Coverage for src/gitlabracadabra/containers/registry_session.py: 98%

1# 

2# Copyright (C) 2019-2025 Mathieu Parent <math.parent@gmail.com> 

3# 

4# This program is free software: you can redistribute it and/or modify 

5# it under the terms of the GNU Lesser General Public License as published by 

6# the Free Software Foundation, either version 3 of the License, or 

7# (at your option) any later version. 

8# 

9# This program is distributed in the hope that it will be useful, 

10# but WITHOUT ANY WARRANTY; without even the implied warranty of 

11# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the 

12# GNU Lesser General Public License for more details. 

13# 

14# You should have received a copy of the GNU Lesser General Public License 

15# along with this program. If not, see <http://www.gnu.org/licenses/>. 

16 

17from __future__ import annotations 

18 

19from typing import TYPE_CHECKING 

20 

21from gitlabracadabra.containers.authenticated_session import AuthenticatedSession 

22from gitlabracadabra.containers.const import DOCKER_HOSTNAME, DOCKER_REGISTRY 

23 

24if TYPE_CHECKING: 

25 from collections.abc import Callable 

26 

27 from requests import Response 

28 from requests.auth import AuthBase 

29 

30 from gitlabracadabra.containers.authenticated_session import Data, Params 

31 from gitlabracadabra.containers.scope import Scope 

32 

33 

34class RegistrySession: 

35 """Container registry HTTP methods.""" 

36 

37 def __init__( 

38 self, 

39 hostname: str, 

40 session_callback: Callable[[AuthenticatedSession], None] | None = None, 

41 ) -> None: 

42 """Instantiate a registry connection. 

43 

44 Args: 

45 hostname: fqdn of a registry. 

46 session_callback: Callback applied to requests Session. 

47 """ 

48 self._session = AuthenticatedSession() 

49 self._hostname = hostname 

50 if hostname == DOCKER_HOSTNAME: 

51 self._session.connection_hostname = DOCKER_REGISTRY 

52 else: 

53 self._session.connection_hostname = hostname 

54 if session_callback is not None: 

55 session_callback(self._session) 

56 # Cache where blobs are present 

57 # dict key is digest, value is a list of manifest names 

58 # Used in WithBlobs 

59 self._blobs: dict[str, list[str]] = {} 

60 self._sizes: dict[str, int] = {} 

61 

62 def __del__(self) -> None: 

63 """Destroy a registry connection.""" 

64 self._session.close() 

65 

66 @property 

67 def hostname(self) -> str: 

68 """Get hostname. 

69 

70 Returns: 

71 The registry hostname. 

72 """ 

73 return self._hostname 

74 

75 def request( 

76 self, 

77 method: str, 

78 url: str, 

79 *, 

80 scopes: set[Scope] | None = None, 

81 params: Params = None, 

82 data: Data | None = None, 

83 headers: dict[str, str] | None = None, 

84 content_type: str | None = None, 

85 accept: tuple[str, ...] | None = None, 

86 auth: AuthBase | None = None, 

87 stream: bool | None = None, 

88 raise_for_status: bool = True, 

89 ) -> Response: 

90 """Send an HTTP request. 

91 

92 Args: 

93 method: HTTP method. 

94 url: Either a path or a full url. 

95 scopes: An optional set of scopes. 

96 params: query string params. 

97 data: Request body stream. 

98 headers: Request headers. 

99 content_type: Uploaded MIME type. 

100 accept: An optional list of accepted mime-types. 

101 auth: HTTPBasicAuth. 

102 stream: Stream the response. 

103 raise_for_status: Raises `requests.HTTPError`, if one occurred. 

104 

105 Returns: 

106 A Response. 

107 """ 

108 headers = headers.copy() if headers else {} 

109 if accept: 

110 headers["Accept"] = ", ".join(accept) 

111 if content_type: 

112 headers["Content-Type"] = content_type 

113 

114 self._session.connect(scopes) 

115 response = self._session.authenticated_request( 

116 method, 

117 url, 

118 params=params, 

119 data=data, 

120 headers=headers, 

121 auth=auth, 

122 stream=stream, 

123 ) 

124 if raise_for_status: 124 ↛ 126line 124 didn't jump to line 126 because the condition on line 124 was always true

125 response.raise_for_status() 

126 return response