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