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

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

282 statements  

1from __future__ import annotations 

2 

3import collections.abc as cabc 

4import re 

5import typing as t 

6import wsgiref.types as wt 

7 

8from .._internal import _missing 

9from ..exceptions import BadRequestKeyError 

10from ..http import dump_options_header 

11from .mixins import ImmutableHeadersMixin 

12from .structures import iter_multi_items 

13from .structures import MultiDict 

14 

15T = t.TypeVar("T") 

16 

17 

18class Headers: 

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

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

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

22 

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

24 headers which are stored as tuples in a list. 

25 

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

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

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

29 catch-all for HTTP exceptions. 

30 

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

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

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

34 a :class:`KeyError`. 

35 

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

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

38 validated the same way values added later are. 

39 

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

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

42 will raise a ``ValueError``. 

43 

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

45 

46 .. versionchanged:: 3.1 

47 Implement ``|`` and ``|=`` operators. 

48 

49 .. versionchanged:: 2.1.0 

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

51 

52 .. versionchanged:: 0.9 

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

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

55 well which will automatically be latin1 decoded. 

56 

57 .. versionchanged:: 0.9 

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

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

60 """ 

61 

62 def __init__( 

63 self, 

64 defaults: ( 

65 Headers 

66 | MultiDict[str, t.Any] 

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

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

69 | None 

70 ) = None, 

71 ) -> None: 

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

73 

74 if defaults is not None: 

75 self.extend(defaults) 

76 

77 @t.overload 

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

79 @t.overload 

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

81 @t.overload 

82 def __getitem__(self, key: slice) -> t.Self: ... 

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

84 if isinstance(key, str): 

85 return self._get_key(key) 

86 

87 if isinstance(key, int): 

88 return self._list[key] 

89 

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

91 

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

93 ikey = key.lower() 

94 

95 for k, v in self._list: 

96 if k.lower() == ikey: 

97 return v 

98 

99 raise BadRequestKeyError(key) 

100 

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

102 if other.__class__ is not self.__class__: 

103 return NotImplemented 

104 

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

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

107 

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

109 

110 __hash__ = None # type: ignore[assignment] 

111 

112 @t.overload 

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

114 @t.overload 

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

116 @t.overload 

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

118 @t.overload 

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

120 @t.overload 

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

122 def get( # type: ignore[misc] 

123 self, 

124 key: str, 

125 default: str | T | None = None, 

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

127 ) -> str | T | None: 

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

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

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

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

132 found: 

133 

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

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

136 42 

137 

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

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

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

141 returned. 

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

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

144 by this callable the default value is returned. 

145 

146 .. versionchanged:: 3.0 

147 The ``as_bytes`` parameter was removed. 

148 

149 .. versionchanged:: 0.9 

150 The ``as_bytes`` parameter was added. 

151 """ 

152 try: 

153 rv = self._get_key(key) 

154 except KeyError: 

155 return default 

156 

157 if type is None: 

158 return rv 

159 

160 try: 

161 return type(rv) 

162 except ValueError: 

163 return default 

164 

165 @t.overload 

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

167 @t.overload 

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

169 def getlist( 

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

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

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

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

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

175 be converted with the callable defined there. 

176 

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

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

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

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

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

182 

183 .. versionchanged:: 3.0 

184 The ``as_bytes`` parameter was removed. 

185 

186 .. versionchanged:: 0.9 

187 The ``as_bytes`` parameter was added. 

188 """ 

189 ikey = key.lower() 

190 

191 if type is not None: 

192 result = [] 

193 

194 for k, v in self: 

195 if k.lower() == ikey: 

196 try: 

197 result.append(type(v)) 

198 except ValueError: 

199 continue 

200 

201 return result 

202 

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

204 

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

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

207 

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

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

210 """ 

211 return self.getlist(name) 

212 

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

214 for key, value in self: 

215 if lower: 

216 key = key.lower() 

217 yield key, value 

218 

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

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

221 yield key 

222 

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

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

225 yield value 

226 

227 def extend( 

228 self, 

229 arg: ( 

230 Headers 

231 | MultiDict[str, t.Any] 

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

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

234 | None 

235 ) = None, 

236 /, 

237 **kwargs: str, 

238 ) -> None: 

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

240 containing header items as well as keyword arguments. 

241 

242 To replace existing keys instead of extending, use 

243 :meth:`update` instead. 

244 

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

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

247 pairs. 

248 

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

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

251 will raise a ``ValueError``. 

252 

253 .. versionchanged:: 1.0 

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

255 """ 

256 if arg is not None: 

257 for key, value in iter_multi_items(arg): 

258 self.add(key, value) 

259 

260 for key, value in iter_multi_items(kwargs): 

261 self.add(key, value) 

262 

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

264 if isinstance(key, str): 

265 self._del_key(key) 

266 return 

267 

268 del self._list[key] 

269 

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

271 key = key.lower() 

272 new = [] 

273 

274 for k, v in self._list: 

275 if k.lower() != key: 

276 new.append((k, v)) 

277 

278 self._list[:] = new 

279 

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

281 """Remove a key. 

282 

283 :param key: The key to be removed. 

284 """ 

285 return self._del_key(key) 

286 

287 @t.overload 

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

289 @t.overload 

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

291 @t.overload 

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

293 @t.overload 

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

295 @t.overload 

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

297 def pop( 

298 self, 

299 key: str | int | None = None, 

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

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

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

303 

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

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

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

307 item is removed. 

308 :return: an item. 

309 """ 

310 if key is None: 

311 return self._list.pop() 

312 

313 if isinstance(key, int): 

314 return self._list.pop(key) 

315 

316 try: 

317 rv = self._get_key(key) 

318 except KeyError: 

319 if default is not _missing: 

320 return default 

321 

322 raise 

323 

324 self.remove(key) 

325 return rv 

326 

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

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

329 return self._list.pop() 

330 

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

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

333 try: 

334 self._get_key(key) 

335 except KeyError: 

336 return False 

337 

338 return True 

339 

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

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

342 return iter(self._list) 

343 

344 def __len__(self) -> int: 

345 return len(self._list) 

346 

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

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

349 

350 Keyword arguments can specify additional parameters for the header 

351 value, with underscores converted to dashes:: 

352 

353 >>> d = Headers() 

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

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

356 

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

358 behind the scenes. 

359 

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

361 must not come from untrusted user input. 

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

363 a ``ValueError``. 

364 

365 .. versionchanged:: 0.4.1 

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

367 """ 

368 if kwargs: 

369 value = _options_header_vkw(value, kwargs) 

370 

371 value_str = _str_header_value(value) 

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

373 

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

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

376 

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

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

379 """ 

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

381 

382 def clear(self) -> None: 

383 """Clears all headers.""" 

384 self._list.clear() 

385 

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

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

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

389 entry or replaces the first one. 

390 

391 Keyword arguments can specify additional parameters for the header 

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

393 more information. 

394 

395 .. versionchanged:: 0.6.1 

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

397 

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

399 must not come from untrusted user input. 

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

401 a ``ValueError``. 

402 """ 

403 if kwargs: 

404 value = _options_header_vkw(value, kwargs) 

405 

406 value_str = _str_header_value(value) 

407 

408 if not self._list: 

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

410 return 

411 

412 iter_list = iter(self._list) 

413 ikey = key.lower() 

414 

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

416 if old_key.lower() == ikey: 

417 # replace first occurrence 

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

419 break 

420 else: 

421 # no existing occurrences 

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

423 return 

424 

425 # remove remaining occurrences 

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

427 

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

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

430 

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

432 must not come from untrusted user input. 

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

434 will raise a ``ValueError``. 

435 

436 .. versionadded:: 1.0 

437 """ 

438 if values: 

439 values_iter = iter(values) 

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

441 

442 for value in values_iter: 

443 self.add(key, value) 

444 else: 

445 self.remove(key) 

446 

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

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

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

450 return that. 

451 

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

453 must not come from untrusted user input. 

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

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

456 """ 

457 try: 

458 return self._get_key(key) 

459 except KeyError: 

460 pass 

461 

462 self.set(key, default) 

463 return self._get_key(key) 

464 

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

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

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

468 by ``default`` and return that. 

469 

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

471 list will not affect the headers. 

472 

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

474 must not come from untrusted user input. 

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

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

477 

478 .. versionadded:: 1.0 

479 """ 

480 if key not in self: 

481 self.setlist(key, default) 

482 

483 return self.getlist(key) 

484 

485 @t.overload 

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

487 @t.overload 

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

489 @t.overload 

490 def __setitem__( 

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

492 ) -> None: ... 

493 def __setitem__( 

494 self, 

495 key: str | int | slice, 

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

497 ) -> None: 

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

499 if isinstance(key, str): 

500 self.set(key, value) 

501 elif isinstance(key, int): 

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

503 else: 

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

505 

506 def update( 

507 self, 

508 arg: ( 

509 Headers 

510 | MultiDict[str, t.Any] 

511 | cabc.Mapping[ 

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

513 ] 

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

515 | None 

516 ) = None, 

517 /, 

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

519 ) -> None: 

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

521 headers object and keyword arguments. 

522 

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

524 instead. 

525 

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

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

528 pairs. 

529 

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

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

532 will raise a ``ValueError``. 

533 

534 .. versionadded:: 1.0 

535 """ 

536 if arg is not None: 

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

538 for key in arg.keys(): 

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

540 elif isinstance(arg, cabc.Mapping): 

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

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

543 self.setlist(key, value) 

544 else: 

545 self.set(key, value) 

546 else: 

547 for key, value in arg: 

548 self.set(key, value) 

549 

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

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

552 self.setlist(key, value) 

553 else: 

554 self.set(key, value) 

555 

556 def __or__( 

557 self, 

558 other: cabc.Mapping[ 

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

560 ], 

561 ) -> t.Self: 

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

563 return NotImplemented 

564 

565 rv = self.copy() 

566 rv.update(other) 

567 return rv 

568 

569 def __ior__( 

570 self, 

571 other: ( 

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

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

574 ), 

575 ) -> t.Self: 

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

577 return NotImplemented 

578 

579 self.update(other) 

580 return self 

581 

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

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

584 

585 :return: list 

586 """ 

587 return list(self) 

588 

589 def copy(self) -> t.Self: 

590 return self.__class__(self._list) 

591 

592 def __copy__(self) -> t.Self: 

593 return self.copy() 

594 

595 def __str__(self) -> str: 

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

597 strs = [] 

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

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

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

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

602 

603 def __repr__(self) -> str: 

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

605 

606 

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

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

609 

610 

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

612 

613 

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

615 if not isinstance(value, str): 

616 value = str(value) 

617 

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

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

620 

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

622 

623 

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

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

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

627 a WSGI environment. 

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

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

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

631 HTTP exceptions. 

632 """ 

633 

634 def __init__(self, environ: wt.WSGIEnvironment) -> None: 

635 super().__init__() 

636 self.environ = environ 

637 

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

639 if not isinstance(other, EnvironHeaders): 

640 return NotImplemented 

641 

642 return self.environ is other.environ 

643 

644 __hash__ = None 

645 

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

647 return self._get_key(key) 

648 

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

650 if not isinstance(key, str): 

651 raise BadRequestKeyError(key) 

652 

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

654 

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

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

657 

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

659 

660 def __len__(self) -> int: 

661 return sum(1 for _ in self) 

662 

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

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

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

666 "HTTP_CONTENT_TYPE", 

667 "HTTP_CONTENT_LENGTH", 

668 }: 

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

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

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

672 

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

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

675 

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

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