Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/werkzeug/datastructures/headers.py: 35%

Shortcuts on this page

r m x   toggle line displays

j k   next/prev highlighted chunk

0   (zero) top of page

1   (one) first highlighted chunk

284 statements  

1from __future__ import annotations 

2 

3import collections.abc as cabc 

4import re 

5import typing as t 

6 

7from .._internal import _missing 

8from ..exceptions import BadRequestKeyError 

9from .mixins import ImmutableHeadersMixin 

10from .structures import iter_multi_items 

11from .structures import MultiDict 

12 

13if t.TYPE_CHECKING: 

14 import typing_extensions as te 

15 from _typeshed.wsgi import WSGIEnvironment 

16 

17T = t.TypeVar("T") 

18 

19 

20class Headers: 

21 """An object that stores some headers. It has a dict-like interface, 

22 but is ordered, can store the same key multiple times, and iterating 

23 yields ``(key, value)`` pairs instead of only keys. 

24 

25 This data structure is useful if you want a nicer way to handle WSGI 

26 headers which are stored as tuples in a list. 

27 

28 From Werkzeug 0.3 onwards, the :exc:`KeyError` raised by this class is 

29 also a subclass of the :class:`~exceptions.BadRequest` HTTP exception 

30 and will render a page for a ``400 BAD REQUEST`` if caught in a 

31 catch-all for HTTP exceptions. 

32 

33 Headers is mostly compatible with the Python :class:`wsgiref.headers.Headers` 

34 class, with the exception of `__getitem__`. :mod:`wsgiref` will return 

35 `None` for ``headers['missing']``, whereas :class:`Headers` will raise 

36 a :class:`KeyError`. 

37 

38 To create a new ``Headers`` object, pass it a list, dict, or 

39 other ``Headers`` object with default values. These values are 

40 validated the same way values added later are. 

41 

42 The header keys are assumed to be trusted and valid, they must not come 

43 from untrusted user input. Newlines in header values are not allowed and 

44 will raise a ``ValueError``. 

45 

46 :param defaults: The list of default values for the :class:`Headers`. 

47 

48 .. versionchanged:: 3.1 

49 Implement ``|`` and ``|=`` operators. 

50 

51 .. versionchanged:: 2.1.0 

52 Default values are validated the same as values added later. 

53 

54 .. versionchanged:: 0.9 

55 This data structure now stores unicode values similar to how the 

56 multi dicts do it. The main difference is that bytes can be set as 

57 well which will automatically be latin1 decoded. 

58 

59 .. versionchanged:: 0.9 

60 The :meth:`linked` function was removed without replacement as it 

61 was an API that does not support the changes to the encoding model. 

62 """ 

63 

64 def __init__( 

65 self, 

66 defaults: ( 

67 Headers 

68 | MultiDict[str, t.Any] 

69 | cabc.Mapping[str, t.Any | list[t.Any] | tuple[t.Any, ...] | set[t.Any]] 

70 | cabc.Iterable[tuple[str, t.Any]] 

71 | None 

72 ) = None, 

73 ) -> None: 

74 self._list: list[tuple[str, str]] = [] 

75 

76 if defaults is not None: 

77 self.extend(defaults) 

78 

79 @t.overload 

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

81 @t.overload 

82 def __getitem__(self, key: int) -> tuple[str, str]: ... 

83 @t.overload 

84 def __getitem__(self, key: slice) -> te.Self: ... 

85 def __getitem__(self, key: str | int | slice) -> str | tuple[str, str] | te.Self: 

86 if isinstance(key, str): 

87 return self._get_key(key) 

88 

89 if isinstance(key, int): 

90 return self._list[key] 

91 

92 return self.__class__(self._list[key]) 

93 

94 def _get_key(self, key: str) -> str: 

95 ikey = key.lower() 

96 

97 for k, v in self._list: 

98 if k.lower() == ikey: 

99 return v 

100 

101 raise BadRequestKeyError(key) 

102 

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

104 if other.__class__ is not self.__class__: 

105 return NotImplemented 

106 

107 def lowered(item: tuple[str, ...]) -> tuple[str, ...]: 

108 return item[0].lower(), *item[1:] 

109 

110 return set(map(lowered, other._list)) == set(map(lowered, self._list)) 

111 

112 __hash__ = None # type: ignore[assignment] 

113 

114 @t.overload 

115 def get(self, key: str) -> str | None: ... 

116 @t.overload 

117 def get(self, key: str, default: str) -> str: ... 

118 @t.overload 

119 def get(self, key: str, default: T) -> str | T: ... 

120 @t.overload 

121 def get(self, key: str, type: cabc.Callable[[str], T]) -> T | None: ... 

122 @t.overload 

123 def get(self, key: str, default: T, type: cabc.Callable[[str], T]) -> T: ... 

124 def get( # type: ignore[misc] 

125 self, 

126 key: str, 

127 default: str | T | None = None, 

128 type: cabc.Callable[[str], T] | None = None, 

129 ) -> str | T | None: 

130 """Return the default value if the requested data doesn't exist. 

131 If `type` is provided and is a callable it should convert the value, 

132 return it or raise a :exc:`ValueError` if that is not possible. In 

133 this case the function will return the default as if the value was not 

134 found: 

135 

136 >>> d = Headers([('Content-Length', '42')]) 

137 >>> d.get('Content-Length', type=int) 

138 42 

139 

140 :param key: The key to be looked up. 

141 :param default: The default value to be returned if the key can't 

142 be looked up. If not further specified `None` is 

143 returned. 

144 :param type: A callable that is used to cast the value in the 

145 :class:`Headers`. If a :exc:`ValueError` is raised 

146 by this callable the default value is returned. 

147 

148 .. versionchanged:: 3.0 

149 The ``as_bytes`` parameter was removed. 

150 

151 .. versionchanged:: 0.9 

152 The ``as_bytes`` parameter was added. 

153 """ 

154 try: 

155 rv = self._get_key(key) 

156 except KeyError: 

157 return default 

158 

159 if type is None: 

160 return rv 

161 

162 try: 

163 return type(rv) 

164 except ValueError: 

165 return default 

166 

167 @t.overload 

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

169 @t.overload 

170 def getlist(self, key: str, type: cabc.Callable[[str], T]) -> list[T]: ... 

171 def getlist( 

172 self, key: str, type: cabc.Callable[[str], T] | None = None 

173 ) -> list[str] | list[T]: 

174 """Return the list of items for a given key. If that key is not in the 

175 :class:`Headers`, the return value will be an empty list. Just like 

176 :meth:`get`, :meth:`getlist` accepts a `type` parameter. All items will 

177 be converted with the callable defined there. 

178 

179 :param key: The key to be looked up. 

180 :param type: A callable that is used to cast the value in the 

181 :class:`Headers`. If a :exc:`ValueError` is raised 

182 by this callable the value will be removed from the list. 

183 :return: a :class:`list` of all the values for the key. 

184 

185 .. versionchanged:: 3.0 

186 The ``as_bytes`` parameter was removed. 

187 

188 .. versionchanged:: 0.9 

189 The ``as_bytes`` parameter was added. 

190 """ 

191 ikey = key.lower() 

192 

193 if type is not None: 

194 result = [] 

195 

196 for k, v in self: 

197 if k.lower() == ikey: 

198 try: 

199 result.append(type(v)) 

200 except ValueError: 

201 continue 

202 

203 return result 

204 

205 return [v for k, v in self if k.lower() == ikey] 

206 

207 def get_all(self, name: str) -> list[str]: 

208 """Return a list of all the values for the named field. 

209 

210 This method is compatible with the :mod:`wsgiref` 

211 :meth:`~wsgiref.headers.Headers.get_all` method. 

212 """ 

213 return self.getlist(name) 

214 

215 def items(self, lower: bool = False) -> t.Iterable[tuple[str, str]]: 

216 for key, value in self: 

217 if lower: 

218 key = key.lower() 

219 yield key, value 

220 

221 def keys(self, lower: bool = False) -> t.Iterable[str]: 

222 for key, _ in self.items(lower): 

223 yield key 

224 

225 def values(self) -> t.Iterable[str]: 

226 for _, value in self.items(): 

227 yield value 

228 

229 def extend( 

230 self, 

231 arg: ( 

232 Headers 

233 | MultiDict[str, t.Any] 

234 | cabc.Mapping[str, t.Any | list[t.Any] | tuple[t.Any, ...] | set[t.Any]] 

235 | cabc.Iterable[tuple[str, t.Any]] 

236 | None 

237 ) = None, 

238 /, 

239 **kwargs: str, 

240 ) -> None: 

241 """Extend headers in this object with items from another object 

242 containing header items as well as keyword arguments. 

243 

244 To replace existing keys instead of extending, use 

245 :meth:`update` instead. 

246 

247 If provided, the first argument can be another :class:`Headers` 

248 object, a :class:`MultiDict`, :class:`dict`, or iterable of 

249 pairs. 

250 

251 The header keys are assumed to be trusted and valid, they must not come 

252 from untrusted user input. Newlines in header values are not allowed and 

253 will raise a ``ValueError``. 

254 

255 .. versionchanged:: 1.0 

256 Support :class:`MultiDict`. Allow passing ``kwargs``. 

257 """ 

258 if arg is not None: 

259 for key, value in iter_multi_items(arg): 

260 self.add(key, value) 

261 

262 for key, value in iter_multi_items(kwargs): 

263 self.add(key, value) 

264 

265 def __delitem__(self, key: str | int | slice) -> None: 

266 if isinstance(key, str): 

267 self._del_key(key) 

268 return 

269 

270 del self._list[key] 

271 

272 def _del_key(self, key: str) -> None: 

273 key = key.lower() 

274 new = [] 

275 

276 for k, v in self._list: 

277 if k.lower() != key: 

278 new.append((k, v)) 

279 

280 self._list[:] = new 

281 

282 def remove(self, key: str) -> None: 

283 """Remove a key. 

284 

285 :param key: The key to be removed. 

286 """ 

287 return self._del_key(key) 

288 

289 @t.overload 

290 def pop(self) -> tuple[str, str]: ... 

291 @t.overload 

292 def pop(self, key: str) -> str: ... 

293 @t.overload 

294 def pop(self, key: int | None = ...) -> tuple[str, str]: ... 

295 @t.overload 

296 def pop(self, key: str, default: str) -> str: ... 

297 @t.overload 

298 def pop(self, key: str, default: T) -> str | T: ... 

299 def pop( 

300 self, 

301 key: str | int | None = None, 

302 default: str | T = _missing, # type: ignore[assignment] 

303 ) -> str | tuple[str, str] | T: 

304 """Removes and returns a key or index. 

305 

306 :param key: The key to be popped. If this is an integer the item at 

307 that position is removed, if it's a string the value for 

308 that key is. If the key is omitted or `None` the last 

309 item is removed. 

310 :return: an item. 

311 """ 

312 if key is None: 

313 return self._list.pop() 

314 

315 if isinstance(key, int): 

316 return self._list.pop(key) 

317 

318 try: 

319 rv = self._get_key(key) 

320 except KeyError: 

321 if default is not _missing: 

322 return default 

323 

324 raise 

325 

326 self.remove(key) 

327 return rv 

328 

329 def popitem(self) -> tuple[str, str]: 

330 """Removes a key or index and returns a (key, value) item.""" 

331 return self._list.pop() 

332 

333 def __contains__(self, key: str) -> bool: 

334 """Check if a key is present.""" 

335 try: 

336 self._get_key(key) 

337 except KeyError: 

338 return False 

339 

340 return True 

341 

342 def __iter__(self) -> t.Iterator[tuple[str, str]]: 

343 """Yield ``(key, value)`` tuples.""" 

344 return iter(self._list) 

345 

346 def __len__(self) -> int: 

347 return len(self._list) 

348 

349 def add(self, key: str, value: t.Any, /, **kwargs: t.Any) -> None: 

350 """Add a new header tuple to the list. 

351 

352 Keyword arguments can specify additional parameters for the header 

353 value, with underscores converted to dashes:: 

354 

355 >>> d = Headers() 

356 >>> d.add('Content-Type', 'text/plain') 

357 >>> d.add('Content-Disposition', 'attachment', filename='foo.png') 

358 

359 The keyword argument dumping uses :func:`dump_options_header` 

360 behind the scenes. 

361 

362 :param key: The header key. This is assumed to be trusted and valid, it 

363 must not come from untrusted user input. 

364 :param value: The header value. Newlines are not allowed and will raise 

365 a ``ValueError``. 

366 

367 .. versionchanged:: 0.4.1 

368 keyword arguments were added for :mod:`wsgiref` compatibility. 

369 """ 

370 if kwargs: 

371 value = _options_header_vkw(value, kwargs) 

372 

373 value_str = _str_header_value(value) 

374 self._list.append((key, value_str)) 

375 

376 def add_header(self, key: str, value: t.Any, /, **kwargs: t.Any) -> None: 

377 """Add a new header tuple to the list. 

378 

379 An alias for :meth:`add` for compatibility with the :mod:`wsgiref` 

380 :meth:`~wsgiref.headers.Headers.add_header` method. 

381 """ 

382 self.add(key, value, **kwargs) 

383 

384 def clear(self) -> None: 

385 """Clears all headers.""" 

386 self._list.clear() 

387 

388 def set(self, key: str, value: t.Any, /, **kwargs: t.Any) -> None: 

389 """Remove all header tuples for `key` and add a new one. The newly 

390 added key either appears at the end of the list if there was no 

391 entry or replaces the first one. 

392 

393 Keyword arguments can specify additional parameters for the header 

394 value, with underscores converted to dashes. See :meth:`add` for 

395 more information. 

396 

397 .. versionchanged:: 0.6.1 

398 :meth:`set` now accepts the same arguments as :meth:`add`. 

399 

400 :param key: The header key. This is assumed to be trusted and valid, it 

401 must not come from untrusted user input. 

402 :param value: The header value. Newlines are not allowed and will raise 

403 a ``ValueError``. 

404 """ 

405 if kwargs: 

406 value = _options_header_vkw(value, kwargs) 

407 

408 value_str = _str_header_value(value) 

409 

410 if not self._list: 

411 self._list.append((key, value_str)) 

412 return 

413 

414 iter_list = iter(self._list) 

415 ikey = key.lower() 

416 

417 for idx, (old_key, _) in enumerate(iter_list): 

418 if old_key.lower() == ikey: 

419 # replace first occurrence 

420 self._list[idx] = (key, value_str) 

421 break 

422 else: 

423 # no existing occurrences 

424 self._list.append((key, value_str)) 

425 return 

426 

427 # remove remaining occurrences 

428 self._list[idx + 1 :] = [t for t in iter_list if t[0].lower() != ikey] 

429 

430 def setlist(self, key: str, values: cabc.Iterable[t.Any]) -> None: 

431 """Remove any existing values for a header and add new ones. 

432 

433 :param key: The header key. This is assumed to be trusted and valid, it 

434 must not come from untrusted user input. 

435 :param value: An iterable of header values. Newlines are not allowed and 

436 will raise a ``ValueError``. 

437 

438 .. versionadded:: 1.0 

439 """ 

440 if values: 

441 values_iter = iter(values) 

442 self.set(key, next(values_iter)) 

443 

444 for value in values_iter: 

445 self.add(key, value) 

446 else: 

447 self.remove(key) 

448 

449 def setdefault(self, key: str, default: t.Any) -> str: 

450 """Return the first value for the key if it is in the headers, 

451 otherwise set the header to the value given by ``default`` and 

452 return that. 

453 

454 :param key: The header key. This is assumed to be trusted and valid, it 

455 must not come from untrusted user input. 

456 :param default: The header value to set if the key is not present. 

457 Newlines are not allowed and will raise a ``ValueError``. 

458 """ 

459 try: 

460 return self._get_key(key) 

461 except KeyError: 

462 pass 

463 

464 self.set(key, default) 

465 return self._get_key(key) 

466 

467 def setlistdefault(self, key: str, default: cabc.Iterable[t.Any]) -> list[str]: 

468 """Return the list of values for the key if it is in the 

469 headers, otherwise set the header to the list of values given 

470 by ``default`` and return that. 

471 

472 Unlike :meth:`MultiDict.setlistdefault`, modifying the returned 

473 list will not affect the headers. 

474 

475 :param key: The header key. This is assumed to be trusted and valid, it 

476 must not come from untrusted user input. 

477 :param default: An iterable of header values to set if the key is not 

478 present. Newlines are not allowed and will raise a ``ValueError``. 

479 

480 .. versionadded:: 1.0 

481 """ 

482 if key not in self: 

483 self.setlist(key, default) 

484 

485 return self.getlist(key) 

486 

487 @t.overload 

488 def __setitem__(self, key: str, value: t.Any) -> None: ... 

489 @t.overload 

490 def __setitem__(self, key: int, value: tuple[str, t.Any]) -> None: ... 

491 @t.overload 

492 def __setitem__( 

493 self, key: slice, value: cabc.Iterable[tuple[str, t.Any]] 

494 ) -> None: ... 

495 def __setitem__( 

496 self, 

497 key: str | int | slice, 

498 value: t.Any | tuple[str, t.Any] | cabc.Iterable[tuple[str, t.Any]], 

499 ) -> None: 

500 """Like :meth:`set` but also supports index/slice based setting.""" 

501 if isinstance(key, str): 

502 self.set(key, value) 

503 elif isinstance(key, int): 

504 self._list[key] = value[0], _str_header_value(value[1]) # type: ignore[index] 

505 else: 

506 self._list[key] = [(k, _str_header_value(v)) for k, v in value] # type: ignore[str-unpack] 

507 

508 def update( 

509 self, 

510 arg: ( 

511 Headers 

512 | MultiDict[str, t.Any] 

513 | cabc.Mapping[ 

514 str, t.Any | list[t.Any] | tuple[t.Any, ...] | cabc.Set[t.Any] 

515 ] 

516 | cabc.Iterable[tuple[str, t.Any]] 

517 | None 

518 ) = None, 

519 /, 

520 **kwargs: t.Any | list[t.Any] | tuple[t.Any, ...] | cabc.Set[t.Any], 

521 ) -> None: 

522 """Replace headers in this object with items from another 

523 headers object and keyword arguments. 

524 

525 To extend existing keys instead of replacing, use :meth:`extend` 

526 instead. 

527 

528 If provided, the first argument can be another :class:`Headers` 

529 object, a :class:`MultiDict`, :class:`dict`, or iterable of 

530 pairs. 

531 

532 The header keys are assumed to be trusted and valid, they must not come 

533 from untrusted user input. Newlines in header values are not allowed and 

534 will raise a ``ValueError``. 

535 

536 .. versionadded:: 1.0 

537 """ 

538 if arg is not None: 

539 if isinstance(arg, (Headers, MultiDict)): 

540 for key in arg.keys(): 

541 self.setlist(key, arg.getlist(key)) 

542 elif isinstance(arg, cabc.Mapping): 

543 for key, value in arg.items(): 

544 if isinstance(value, (list, tuple, set)): 

545 self.setlist(key, value) 

546 else: 

547 self.set(key, value) 

548 else: 

549 for key, value in arg: 

550 self.set(key, value) 

551 

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

553 if isinstance(value, (list, tuple, set)): 

554 self.setlist(key, value) 

555 else: 

556 self.set(key, value) 

557 

558 def __or__( 

559 self, 

560 other: cabc.Mapping[ 

561 str, t.Any | list[t.Any] | tuple[t.Any, ...] | cabc.Set[t.Any] 

562 ], 

563 ) -> te.Self: 

564 if not isinstance(other, cabc.Mapping): 

565 return NotImplemented 

566 

567 rv = self.copy() 

568 rv.update(other) 

569 return rv 

570 

571 def __ior__( 

572 self, 

573 other: ( 

574 cabc.Mapping[str, t.Any | list[t.Any] | tuple[t.Any, ...] | cabc.Set[t.Any]] 

575 | cabc.Iterable[tuple[str, t.Any]] 

576 ), 

577 ) -> te.Self: 

578 if not isinstance(other, (cabc.Mapping, cabc.Iterable)): 

579 return NotImplemented 

580 

581 self.update(other) 

582 return self 

583 

584 def to_wsgi_list(self) -> list[tuple[str, str]]: 

585 """Convert the headers into a list suitable for WSGI. 

586 

587 :return: list 

588 """ 

589 return list(self) 

590 

591 def copy(self) -> te.Self: 

592 return self.__class__(self._list) 

593 

594 def __copy__(self) -> te.Self: 

595 return self.copy() 

596 

597 def __str__(self) -> str: 

598 """Returns formatted headers suitable for HTTP transmission.""" 

599 strs = [] 

600 for key, value in self.to_wsgi_list(): 

601 strs.append(f"{key}: {value}") 

602 strs.append("\r\n") 

603 return "\r\n".join(strs) 

604 

605 def __repr__(self) -> str: 

606 return f"{type(self).__name__}({list(self)!r})" 

607 

608 

609def _options_header_vkw(value: str, kw: dict[str, t.Any]) -> str: 

610 return http.dump_options_header( 

611 value, {k.replace("_", "-"): v for k, v in kw.items()} 

612 ) 

613 

614 

615_newline_re = re.compile(r"[\r\n]") 

616 

617 

618def _str_header_value(value: t.Any) -> str: 

619 if not isinstance(value, str): 

620 value = str(value) 

621 

622 if _newline_re.search(value) is not None: 

623 raise ValueError("Header values must not contain newline characters.") 

624 

625 return value # type: ignore[no-any-return] 

626 

627 

628class EnvironHeaders(ImmutableHeadersMixin, Headers): # type: ignore[misc] 

629 """Read only version of the headers from a WSGI environment. This 

630 provides the same interface as `Headers` and is constructed from 

631 a WSGI environment. 

632 From Werkzeug 0.3 onwards, the `KeyError` raised by this class is also a 

633 subclass of the :exc:`~exceptions.BadRequest` HTTP exception and will 

634 render a page for a ``400 BAD REQUEST`` if caught in a catch-all for 

635 HTTP exceptions. 

636 """ 

637 

638 def __init__(self, environ: WSGIEnvironment) -> None: 

639 super().__init__() 

640 self.environ = environ 

641 

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

643 if not isinstance(other, EnvironHeaders): 

644 return NotImplemented 

645 

646 return self.environ is other.environ 

647 

648 __hash__ = None 

649 

650 def __getitem__(self, key: str) -> str: # type: ignore[override] 

651 return self._get_key(key) 

652 

653 def _get_key(self, key: str) -> str: 

654 if not isinstance(key, str): 

655 raise BadRequestKeyError(key) 

656 

657 key = key.upper().replace("-", "_") 

658 

659 if key in {"CONTENT_TYPE", "CONTENT_LENGTH"}: 

660 return self.environ[key] # type: ignore[no-any-return] 

661 

662 return self.environ[f"HTTP_{key}"] # type: ignore[no-any-return] 

663 

664 def __len__(self) -> int: 

665 return sum(1 for _ in self) 

666 

667 def __iter__(self) -> cabc.Iterator[tuple[str, str]]: 

668 for key, value in self.environ.items(): 

669 if key.startswith("HTTP_") and key not in { 

670 "HTTP_CONTENT_TYPE", 

671 "HTTP_CONTENT_LENGTH", 

672 }: 

673 yield key[5:].replace("_", "-").title(), value 

674 elif key in {"CONTENT_TYPE", "CONTENT_LENGTH"} and value: 

675 yield key.replace("_", "-").title(), value 

676 

677 def copy(self) -> t.NoReturn: 

678 raise TypeError(f"cannot create {type(self).__name__!r} copies") 

679 

680 def __or__(self, other: t.Any) -> t.NoReturn: 

681 raise TypeError(f"cannot create {type(self).__name__!r} copies") 

682 

683 

684# circular dependencies 

685from .. import http # noqa: E402