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")