Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/urllib3/_collections.py: 31%

1from __future__ import annotations 

2 

3import typing 

4from collections import OrderedDict 

5from enum import Enum, auto 

6from threading import RLock 

7 

8if typing.TYPE_CHECKING: 

9 # We can only import Protocol if TYPE_CHECKING because it's a development 

10 # dependency, and is not available at runtime. 

11 from typing import Protocol 

12 

13 from typing_extensions import Self 

14 

15 class HasGettableStringKeys(Protocol): 

16 def keys(self) -> typing.Iterator[str]: ... 

17 

18 def __getitem__(self, key: str) -> str: ... 

19 

20 

21__all__ = ["RecentlyUsedContainer", "HTTPHeaderDict"] 

22 

23 

24# Key type 

25_KT = typing.TypeVar("_KT") 

26# Value type 

27_VT = typing.TypeVar("_VT") 

28# Default type 

29_DT = typing.TypeVar("_DT") 

30 

31ValidHTTPHeaderSource = typing.Union[ 

32 "HTTPHeaderDict", 

33 typing.Mapping[str, str], 

34 typing.Iterable[tuple[str, str]], 

35 "HasGettableStringKeys", 

36] 

37 

38 

39class _Sentinel(Enum): 

40 not_passed = auto() 

41 

42 

43def ensure_can_construct_http_header_dict( 

44 potential: object, 

45) -> ValidHTTPHeaderSource | None: 

46 if isinstance(potential, HTTPHeaderDict): 

47 return potential 

48 elif isinstance(potential, typing.Mapping): 

49 # Full runtime checking of the contents of a Mapping is expensive, so for the 

50 # purposes of typechecking, we assume that any Mapping is the right shape. 

51 return typing.cast(typing.Mapping[str, str], potential) 

52 elif isinstance(potential, typing.Iterable): 

53 # Similarly to Mapping, full runtime checking of the contents of an Iterable is 

54 # expensive, so for the purposes of typechecking, we assume that any Iterable 

55 # is the right shape. 

56 return typing.cast(typing.Iterable[tuple[str, str]], potential) 

57 elif hasattr(potential, "keys") and hasattr(potential, "__getitem__"): 

58 return typing.cast("HasGettableStringKeys", potential) 

59 else: 

60 return None 

61 

62 

63class RecentlyUsedContainer(typing.Generic[_KT, _VT], typing.MutableMapping[_KT, _VT]): 

64 """ 

65 Provides a thread-safe dict-like container which maintains up to 

66 ``maxsize`` keys while throwing away the least-recently-used keys beyond 

67 ``maxsize``. 

68 

69 :param maxsize: 

70 Maximum number of recent elements to retain. 

71 

72 :param dispose_func: 

73 Every time an item is evicted from the container, 

74 ``dispose_func(value)`` is called. Callback which will get called 

75 """ 

76 

77 _container: typing.OrderedDict[_KT, _VT] 

78 _maxsize: int 

79 dispose_func: typing.Callable[[_VT], None] | None 

80 lock: RLock 

81 

82 def __init__( 

83 self, 

84 maxsize: int = 10, 

85 dispose_func: typing.Callable[[_VT], None] | None = None, 

86 ) -> None: 

87 super().__init__() 

88 self._maxsize = maxsize 

89 self.dispose_func = dispose_func 

90 self._container = OrderedDict() 

91 self.lock = RLock() 

92 

93 def __getitem__(self, key: _KT) -> _VT: 

94 # Move the item to the end of the eviction line. move_to_end() keeps 

95 # the stored key instead of replacing it with the caller's key. 

96 with self.lock: 

97 item = self._container[key] 

98 self._container.move_to_end(key) 

99 return item 

100 

101 def __setitem__(self, key: _KT, value: _VT) -> None: 

102 evicted_item = None 

103 with self.lock: 

104 # Possibly evict the existing value of 'key' 

105 try: 

106 # If the key exists, we'll overwrite it, which won't change the 

107 # size of the pool. Because accessing a key should move it to 

108 # the end of the eviction line, we pop it out first. 

109 evicted_item = key, self._container.pop(key) 

110 self._container[key] = value 

111 except KeyError: 

112 # When the key does not exist, we insert the value first so that 

113 # evicting works in all cases, including when self._maxsize is 0 

114 self._container[key] = value 

115 if len(self._container) > self._maxsize: 

116 # If we didn't evict an existing value, and we've hit our maximum 

117 # size, then we have to evict the least recently used item from 

118 # the beginning of the container. 

119 evicted_item = self._container.popitem(last=False) 

120 

121 # After releasing the lock on the pool, dispose of any evicted value. 

122 if evicted_item is not None and self.dispose_func: 

123 _, evicted_value = evicted_item 

124 self.dispose_func(evicted_value) 

125 

126 def __delitem__(self, key: _KT) -> None: 

127 with self.lock: 

128 value = self._container.pop(key) 

129 

130 if self.dispose_func: 

131 self.dispose_func(value) 

132 

133 def __len__(self) -> int: 

134 with self.lock: 

135 return len(self._container) 

136 

137 def __iter__(self) -> typing.NoReturn: 

138 raise NotImplementedError( 

139 "Iteration over this class is unlikely to be threadsafe." 

140 ) 

141 

142 def clear(self) -> None: 

143 with self.lock: 

144 # Copy pointers to all values, then wipe the mapping 

145 values = list(self._container.values()) 

146 self._container.clear() 

147 

148 if self.dispose_func: 

149 for value in values: 

150 self.dispose_func(value) 

151 

152 def keys(self) -> set[_KT]: # type: ignore[override] 

153 with self.lock: 

154 return set(self._container.keys()) 

155 

156 

157class HTTPHeaderDictItemView(set[tuple[str, str]]): 

158 """ 

159 HTTPHeaderDict is unusual for a Mapping[str, str] in that it has two modes of 

160 address. 

161 

162 If we directly try to get an item with a particular name, we will get a string 

163 back that is the concatenated version of all the values: 

164 

165 >>> d['X-Header-Name'] 

166 'Value1, Value2, Value3' 

167 

168 However, if we iterate over an HTTPHeaderDict's items, we will optionally combine 

169 these values based on whether combine=True was called when building up the dictionary 

170 

171 >>> d = HTTPHeaderDict({"A": "1", "B": "foo"}) 

172 >>> d.add("A", "2", combine=True) 

173 >>> d.add("B", "bar") 

174 >>> list(d.items()) 

175 [ 

176 ('A', '1, 2'), 

177 ('B', 'foo'), 

178 ('B', 'bar'), 

179 ] 

180 

181 This class conforms to the interface required by the MutableMapping ABC while 

182 also giving us the nonstandard iteration behavior we want; items with duplicate 

183 keys, ordered by time of first insertion. 

184 """ 

185 

186 _headers: HTTPHeaderDict 

187 

188 def __init__(self, headers: HTTPHeaderDict) -> None: 

189 self._headers = headers 

190 

191 def __len__(self) -> int: 

192 return len(list(self._headers.iteritems())) 

193 

194 def __iter__(self) -> typing.Iterator[tuple[str, str]]: 

195 return self._headers.iteritems() 

196 

197 def __contains__(self, item: object) -> bool: 

198 if isinstance(item, tuple) and len(item) == 2: 

199 passed_key, passed_val = item 

200 if isinstance(passed_key, str) and isinstance(passed_val, str): 

201 return self._headers._has_value_for_header(passed_key, passed_val) 

202 return False 

203 

204 

205class HTTPHeaderDict(typing.MutableMapping[str, str]): 

206 """ 

207 :param headers: 

208 An iterable of field-value pairs. Must not contain multiple field names 

209 when compared case-insensitively. 

210 

211 :param kwargs: 

212 Additional field-value pairs to pass in to ``dict.update``. 

213 

214 A ``dict`` like container for storing HTTP Headers. 

215 

216 Field names are stored and compared case-insensitively in compliance with 

217 RFC 7230. Iteration provides the first case-sensitive key seen for each 

218 case-insensitive pair. 

219 

220 Using ``__setitem__`` syntax overwrites fields that compare equal 

221 case-insensitively in order to maintain ``dict``'s api. For fields that 

222 compare equal, instead create a new ``HTTPHeaderDict`` and use ``.add`` 

223 in a loop. 

224 

225 If multiple fields that are equal case-insensitively are passed to the 

226 constructor or ``.update``, the behavior is undefined and some will be 

227 lost. 

228 

229 >>> headers = HTTPHeaderDict() 

230 >>> headers.add('Set-Cookie', 'foo=bar') 

231 >>> headers.add('set-cookie', 'baz=quxx') 

232 >>> headers['content-length'] = '7' 

233 >>> headers['SET-cookie'] 

234 'foo=bar, baz=quxx' 

235 >>> headers['Content-Length'] 

236 '7' 

237 """ 

238 

239 _container: typing.MutableMapping[str, list[str]] 

240 

241 def __init__(self, headers: ValidHTTPHeaderSource | None = None, **kwargs: str): 

242 super().__init__() 

243 self._container = {} # 'dict' is insert-ordered 

244 if headers is not None: 

245 if isinstance(headers, HTTPHeaderDict): 

246 self._copy_from(headers) 

247 else: 

248 self.extend(headers) 

249 if kwargs: 

250 self.extend(kwargs) 

251 

252 def __setitem__(self, key: str, val: str) -> None: 

253 # avoid a bytes/str comparison by decoding before httplib 

254 if isinstance(key, bytes): 

255 key = key.decode("latin-1") 

256 self._container[key.lower()] = [key, val] 

257 

258 def __getitem__(self, key: str) -> str: 

259 if isinstance(key, bytes): 

260 key = key.decode("latin-1") 

261 val = self._container[key.lower()] 

262 return ", ".join(val[1:]) 

263 

264 def __delitem__(self, key: str) -> None: 

265 if isinstance(key, bytes): 

266 key = key.decode("latin-1") 

267 del self._container[key.lower()] 

268 

269 def __contains__(self, key: object) -> bool: 

270 if isinstance(key, bytes): 

271 key = key.decode("latin-1") 

272 if isinstance(key, str): 

273 return key.lower() in self._container 

274 return False 

275 

276 def setdefault(self, key: str, default: str = "") -> str: 

277 return super().setdefault(key, default) 

278 

279 def __eq__(self, other: object) -> bool: 

280 maybe_constructable = ensure_can_construct_http_header_dict(other) 

281 if maybe_constructable is None: 

282 return False 

283 else: 

284 other_as_http_header_dict = type(self)(maybe_constructable) 

285 

286 return {k.lower(): v for k, v in self.itermerged()} == { 

287 k.lower(): v for k, v in other_as_http_header_dict.itermerged() 

288 } 

289 

290 def __ne__(self, other: object) -> bool: 

291 return not self.__eq__(other) 

292 

293 def __len__(self) -> int: 

294 return len(self._container) 

295 

296 def __iter__(self) -> typing.Iterator[str]: 

297 # Only provide the originally cased names 

298 for vals in self._container.values(): 

299 yield vals[0] 

300 

301 def discard(self, key: str) -> None: 

302 try: 

303 del self[key] 

304 except KeyError: 

305 pass 

306 

307 def add(self, key: str, val: str, *, combine: bool = False) -> None: 

308 """Adds a (name, value) pair, doesn't overwrite the value if it already 

309 exists. 

310 

311 If this is called with combine=True, instead of adding a new header value 

312 as a distinct item during iteration, this will instead append the value to 

313 the last existing header value with a comma. If no existing header value exists 

314 for the key, then the value will simply be added, ignoring the combine parameter. 

315 

316 >>> headers = HTTPHeaderDict(foo='bar') 

317 >>> headers.add('Foo', 'baz') 

318 >>> headers['foo'] 

319 'bar, baz' 

320 >>> list(headers.items()) 

321 [('foo', 'bar'), ('foo', 'baz')] 

322 >>> headers.add('foo', 'quz', combine=True) 

323 >>> list(headers.items()) 

324 [('foo', 'bar'), ('foo', 'baz, quz')] 

325 """ 

326 # avoid a bytes/str comparison by decoding before httplib 

327 if isinstance(key, bytes): 

328 key = key.decode("latin-1") 

329 key_lower = key.lower() 

330 new_vals = [key, val] 

331 # Keep the common case aka no item present as fast as possible 

332 vals = self._container.setdefault(key_lower, new_vals) 

333 if new_vals is not vals: 

334 # if there are values here, then there is at least the initial 

335 # key/value pair 

336 assert len(vals) >= 2 

337 if combine: 

338 vals[-1] = vals[-1] + ", " + val 

339 else: 

340 vals.append(val) 

341 

342 def extend(self, *args: ValidHTTPHeaderSource, **kwargs: str) -> None: 

343 """Generic import function for any type of header-like object. 

344 Adapted version of MutableMapping.update in order to insert items 

345 with self.add instead of self.__setitem__ 

346 """ 

347 if len(args) > 1: 

348 raise TypeError( 

349 f"extend() takes at most 1 positional arguments ({len(args)} given)" 

350 ) 

351 other = args[0] if len(args) >= 1 else () 

352 

353 if isinstance(other, HTTPHeaderDict): 

354 for key, val in other.iteritems(): 

355 self.add(key, val) 

356 elif isinstance(other, typing.Mapping): 

357 for key, val in other.items(): 

358 self.add(key, val) 

359 elif isinstance(other, typing.Iterable): 

360 for key, value in other: 

361 self.add(key, value) 

362 elif hasattr(other, "keys") and hasattr(other, "__getitem__"): 

363 # THIS IS NOT A TYPESAFE BRANCH 

364 # In this branch, the object has a `keys` attr but is not a Mapping or any of 

365 # the other types indicated in the method signature. We do some stuff with 

366 # it as though it partially implements the Mapping interface, but we're not 

367 # doing that stuff safely AT ALL. 

368 for key in other.keys(): 

369 self.add(key, other[key]) 

370 

371 for key, value in kwargs.items(): 

372 self.add(key, value) 

373 

374 @typing.overload 

375 def getlist(self, key: str) -> list[str]: ... 

376 

377 @typing.overload 

378 def getlist(self, key: str, default: _DT) -> list[str] | _DT: ... 

379 

380 def getlist( 

381 self, key: str, default: _Sentinel | _DT = _Sentinel.not_passed 

382 ) -> list[str] | _DT: 

383 """Returns a list of all the values for the named field. Returns an 

384 empty list if the key doesn't exist.""" 

385 if isinstance(key, bytes): 

386 key = key.decode("latin-1") 

387 try: 

388 vals = self._container[key.lower()] 

389 except KeyError: 

390 if default is _Sentinel.not_passed: 

391 # _DT is unbound; empty list is instance of List[str] 

392 return [] 

393 # _DT is bound; default is instance of _DT 

394 return default 

395 else: 

396 # _DT may or may not be bound; vals[1:] is instance of List[str], which 

397 # meets our external interface requirement of `Union[List[str], _DT]`. 

398 return vals[1:] 

399 

400 def _prepare_for_method_change(self) -> Self: 

401 """ 

402 Remove content-specific header fields before changing the request 

403 method to GET or HEAD according to RFC 9110, Section 15.4. 

404 

405 ``Transfer-Encoding`` goes as well: it frames a request body that is 

406 no longer going to be sent. 

407 """ 

408 content_specific_headers = [ 

409 "Content-Encoding", 

410 "Content-Language", 

411 "Content-Location", 

412 "Content-Type", 

413 "Content-Length", 

414 "Digest", 

415 "Last-Modified", 

416 "Transfer-Encoding", 

417 ] 

418 for header in content_specific_headers: 

419 self.discard(header) 

420 return self 

421 

422 # Backwards compatibility for httplib 

423 getheaders = getlist 

424 getallmatchingheaders = getlist 

425 iget = getlist 

426 

427 # Backwards compatibility for http.cookiejar 

428 get_all = getlist 

429 

430 def __repr__(self) -> str: 

431 return f"{type(self).__name__}({dict(self.itermerged())})" 

432 

433 def _copy_from(self, other: HTTPHeaderDict) -> None: 

434 for key in other: 

435 val = other.getlist(key) 

436 self._container[key.lower()] = [key, *val] 

437 

438 def copy(self) -> Self: 

439 clone = type(self)() 

440 clone._copy_from(self) 

441 return clone 

442 

443 def iteritems(self) -> typing.Iterator[tuple[str, str]]: 

444 """Iterate over all header lines, including duplicate ones.""" 

445 for key in self: 

446 vals = self._container[key.lower()] 

447 for val in vals[1:]: 

448 yield vals[0], val 

449 

450 def itermerged(self) -> typing.Iterator[tuple[str, str]]: 

451 """Iterate over all headers, merging duplicate ones together.""" 

452 for key in self: 

453 val = self._container[key.lower()] 

454 yield val[0], ", ".join(val[1:]) 

455 

456 def items(self) -> HTTPHeaderDictItemView: # type: ignore[override] 

457 return HTTPHeaderDictItemView(self) 

458 

459 def _has_value_for_header(self, header_name: str, potential_value: str) -> bool: 

460 if header_name in self: 

461 return potential_value in self._container[header_name.lower()][1:] 

462 return False 

463 

464 def __ior__(self, other: object) -> HTTPHeaderDict: 

465 # Supports extending a header dict in-place using operator |= 

466 # combining items with add instead of __setitem__ 

467 maybe_constructable = ensure_can_construct_http_header_dict(other) 

468 if maybe_constructable is None: 

469 return NotImplemented 

470 self.extend(maybe_constructable) 

471 return self 

472 

473 def __or__(self, other: object) -> Self: 

474 # Supports merging header dicts using operator | 

475 # combining items with add instead of __setitem__ 

476 maybe_constructable = ensure_can_construct_http_header_dict(other) 

477 if maybe_constructable is None: 

478 return NotImplemented 

479 result = self.copy() 

480 result.extend(maybe_constructable) 

481 return result 

482 

483 def __ror__(self, other: object) -> Self: 

484 # Supports merging header dicts using operator | when other is on left side 

485 # combining items with add instead of __setitem__ 

486 maybe_constructable = ensure_can_construct_http_header_dict(other) 

487 if maybe_constructable is None: 

488 return NotImplemented 

489 result = type(self)(maybe_constructable) 

490 result.extend(self) 

491 return result