Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/pip/_vendor/cachecontrol/adapter.py: 24%

1# SPDX-FileCopyrightText: 2015 Eric Larson 

2# 

3# SPDX-License-Identifier: Apache-2.0 

4from __future__ import annotations 

5 

6import functools 

7import types 

8import zlib 

9from typing import TYPE_CHECKING, Any, Collection, Mapping 

10 

11from pip._vendor.requests.adapters import HTTPAdapter 

12 

13from pip._vendor.cachecontrol.cache import DictCache 

14from pip._vendor.cachecontrol.controller import PERMANENT_REDIRECT_STATUSES, CacheController 

15from pip._vendor.cachecontrol.filewrapper import CallbackFileWrapper 

16 

17if TYPE_CHECKING: 

18 from pip._vendor.requests import PreparedRequest, Response 

19 from pip._vendor.urllib3 import HTTPResponse 

20 

21 from pip._vendor.cachecontrol.cache import BaseCache 

22 from pip._vendor.cachecontrol.heuristics import BaseHeuristic 

23 from pip._vendor.cachecontrol.serialize import Serializer 

24 

25 

26class CacheControlAdapter(HTTPAdapter): 

27 invalidating_methods = {"PUT", "PATCH", "DELETE"} 

28 

29 def __init__( 

30 self, 

31 cache: BaseCache | None = None, 

32 cache_etags: bool = True, 

33 controller_class: type[CacheController] | None = None, 

34 serializer: Serializer | None = None, 

35 heuristic: BaseHeuristic | None = None, 

36 cacheable_methods: Collection[str] | None = None, 

37 *args: Any, 

38 **kw: Any, 

39 ) -> None: 

40 super().__init__(*args, **kw) 

41 self.cache = DictCache() if cache is None else cache 

42 self.heuristic = heuristic 

43 self.cacheable_methods = cacheable_methods or ("GET",) 

44 

45 controller_factory = controller_class or CacheController 

46 self.controller = controller_factory( 

47 self.cache, cache_etags=cache_etags, serializer=serializer 

48 ) 

49 

50 def send( 

51 self, 

52 request: PreparedRequest, 

53 stream: bool = False, 

54 timeout: None | float | tuple[float, float] | tuple[float, None] = None, 

55 verify: bool | str = True, 

56 cert: (None | bytes | str | tuple[bytes | str, bytes | str]) = None, 

57 proxies: Mapping[str, str] | None = None, 

58 cacheable_methods: Collection[str] | None = None, 

59 ) -> Response: 

60 """ 

61 Send a request. Use the request information to see if it 

62 exists in the cache and cache the response if we need to and can. 

63 """ 

64 cacheable = cacheable_methods or self.cacheable_methods 

65 if request.method in cacheable: 

66 try: 

67 cached_response = self.controller.cached_request(request) 

68 except zlib.error: 

69 cached_response = None 

70 if cached_response: 

71 return self.build_response(request, cached_response, from_cache=True) 

72 

73 # check for etags and add headers if appropriate 

74 request.headers.update(self.controller.conditional_headers(request)) 

75 

76 resp = super().send(request, stream, timeout, verify, cert, proxies) 

77 

78 return resp 

79 

80 def build_response( 

81 self, 

82 request: PreparedRequest, 

83 response: HTTPResponse, 

84 from_cache: bool = False, 

85 cacheable_methods: Collection[str] | None = None, 

86 ) -> Response: 

87 """ 

88 Build a response by making a request or using the cache. 

89 

90 This will end up calling send and returning a potentially 

91 cached response 

92 """ 

93 cacheable = cacheable_methods or self.cacheable_methods 

94 if not from_cache and request.method in cacheable: 

95 # Check for any heuristics that might update headers 

96 # before trying to cache. 

97 if self.heuristic: 

98 response = self.heuristic.apply(response) 

99 

100 # apply any expiration heuristics 

101 if response.status == 304: 

102 # We must have sent an ETag request. This could mean 

103 # that we've been expired already or that we simply 

104 # have an etag. In either case, we want to try and 

105 # update the cache if that is the case. 

106 cached_response = self.controller.update_cached_response( 

107 request, response 

108 ) 

109 

110 if cached_response is not response: 

111 from_cache = True 

112 

113 # We are done with the server response, read a 

114 # possible response body (compliant servers will 

115 # not return one, but we cannot be 100% sure) and 

116 # release the connection back to the pool. 

117 response.read(decode_content=False) 

118 response.release_conn() 

119 

120 response = cached_response 

121 

122 # We always cache the 301 responses 

123 elif int(response.status) in PERMANENT_REDIRECT_STATUSES: 

124 self.controller.cache_response(request, response) 

125 else: 

126 # Wrap the response file with a wrapper that will cache the 

127 # response when the stream has been consumed. 

128 response._fp = CallbackFileWrapper( # type: ignore[attr-defined] 

129 response._fp, # type: ignore[attr-defined] 

130 functools.partial( 

131 self.controller.cache_response, request, response 

132 ), 

133 ) 

134 if response.chunked: 

135 super_update_chunk_length = response._update_chunk_length # type: ignore[attr-defined] 

136 

137 def _update_chunk_length(self: HTTPResponse) -> None: 

138 super_update_chunk_length() 

139 if self.chunk_left == 0: 

140 self._fp._close() # type: ignore[attr-defined] 

141 

142 response._update_chunk_length = types.MethodType( # type: ignore[attr-defined] 

143 _update_chunk_length, response 

144 ) 

145 

146 resp: Response = super().build_response(request, response) # type: ignore[no-untyped-call] 

147 

148 # See if we should invalidate the cache. 

149 if request.method in self.invalidating_methods and resp.ok: 

150 assert request.url is not None 

151 cache_url = self.controller.cache_url(request.url) 

152 self.cache.delete(cache_url) 

153 

154 # Give the request a from_cache attr to let people use it 

155 resp.from_cache = from_cache # type: ignore[attr-defined] 

156 

157 return resp 

158 

159 def close(self) -> None: 

160 self.cache.close() 

161 super().close() # type: ignore[no-untyped-call]