codekingpro/portable-devtools
114k
1from __future__ import annotations2 3import re4import typing as t5 6from .._internal import _missing7from ..exceptions import BadRequestKeyError8from .mixins import ImmutableHeadersMixin9from .structures import iter_multi_items10from .structures import MultiDict11 12 13class Headers:14 """An object that stores some headers. It has a dict-like interface,15 but is ordered, can store the same key multiple times, and iterating16 yields ``(key, value)`` pairs instead of only keys.17 18 This data structure is useful if you want a nicer way to handle WSGI19 headers which are stored as tuples in a list.20 21 From Werkzeug 0.3 onwards, the :exc:`KeyError` raised by this class is22 also a subclass of the :class:`~exceptions.BadRequest` HTTP exception23 and will render a page for a ``400 BAD REQUEST`` if caught in a24 catch-all for HTTP exceptions.25 26 Headers is mostly compatible with the Python :class:`wsgiref.headers.Headers`27 class, with the exception of `__getitem__`. :mod:`wsgiref` will return28 `None` for ``headers['missing']``, whereas :class:`Headers` will raise29 a :class:`KeyError`.30 31 To create a new ``Headers`` object, pass it a list, dict, or32 other ``Headers`` object with default values. These values are33 validated the same way values added later are.34 35 :param defaults: The list of default values for the :class:`Headers`.36 37 .. versionchanged:: 2.1.038 Default values are validated the same as values added later.39 40 .. versionchanged:: 0.941 This data structure now stores unicode values similar to how the42 multi dicts do it. The main difference is that bytes can be set as43 well which will automatically be latin1 decoded.44 45 .. versionchanged:: 0.946 The :meth:`linked` function was removed without replacement as it47 was an API that does not support the changes to the encoding model.48 """49 50 def __init__(self, defaults=None):51 self._list = []52 if defaults is not None:53 self.extend(defaults)54 55 def __getitem__(self, key, _get_mode=False):56 if not _get_mode:57 if isinstance(key, int):58 return self._list[key]59 elif isinstance(key, slice):60 return self.__class__(self._list[key])61 if not isinstance(key, str):62 raise BadRequestKeyError(key)63 ikey = key.lower()64 for k, v in self._list:65 if k.lower() == ikey:66 return v67 # micro optimization: if we are in get mode we will catch that68 # exception one stack level down so we can raise a standard69 # key error instead of our special one.70 if _get_mode:71 raise KeyError()72 raise BadRequestKeyError(key)73 74 def __eq__(self, other):75 def lowered(item):76 return (item[0].lower(),) + item[1:]77 78 return other.__class__ is self.__class__ and set(79 map(lowered, other._list)80 ) == set(map(lowered, self._list))81 82 __hash__ = None83 84 def get(self, key, default=None, type=None):85 """Return the default value if the requested data doesn't exist.86 If `type` is provided and is a callable it should convert the value,87 return it or raise a :exc:`ValueError` if that is not possible. In88 this case the function will return the default as if the value was not89 found:90 91 >>> d = Headers([('Content-Length', '42')])92 >>> d.get('Content-Length', type=int)93 4294 95 :param key: The key to be looked up.96 :param default: The default value to be returned if the key can't97 be looked up. If not further specified `None` is98 returned.99 :param type: A callable that is used to cast the value in the100 :class:`Headers`. If a :exc:`ValueError` is raised101 by this callable the default value is returned.102 103 .. versionchanged:: 3.0104 The ``as_bytes`` parameter was removed.105 106 .. versionchanged:: 0.9107 The ``as_bytes`` parameter was added.108 """109 try:110 rv = self.__getitem__(key, _get_mode=True)111 except KeyError:112 return default113 if type is None:114 return rv115 try:116 return type(rv)117 except ValueError:118 return default119 120 def getlist(self, key, type=None):121 """Return the list of items for a given key. If that key is not in the122 :class:`Headers`, the return value will be an empty list. Just like123 :meth:`get`, :meth:`getlist` accepts a `type` parameter. All items will124 be converted with the callable defined there.125 126 :param key: The key to be looked up.127 :param type: A callable that is used to cast the value in the128 :class:`Headers`. If a :exc:`ValueError` is raised129 by this callable the value will be removed from the list.130 :return: a :class:`list` of all the values for the key.131 132 .. versionchanged:: 3.0133 The ``as_bytes`` parameter was removed.134 135 .. versionchanged:: 0.9136 The ``as_bytes`` parameter was added.137 """138 ikey = key.lower()139 result = []140 for k, v in self:141 if k.lower() == ikey:142 if type is not None:143 try:144 v = type(v)145 except ValueError:146 continue147 result.append(v)148 return result149 150 def get_all(self, name):151 """Return a list of all the values for the named field.152 153 This method is compatible with the :mod:`wsgiref`154 :meth:`~wsgiref.headers.Headers.get_all` method.155 """156 return self.getlist(name)157 158 def items(self, lower=False):159 for key, value in self:160 if lower:161 key = key.lower()162 yield key, value163 164 def keys(self, lower=False):165 for key, _ in self.items(lower):166 yield key167 168 def values(self):169 for _, value in self.items():170 yield value171 172 def extend(self, *args, **kwargs):173 """Extend headers in this object with items from another object174 containing header items as well as keyword arguments.175 176 To replace existing keys instead of extending, use177 :meth:`update` instead.178 179 If provided, the first argument can be another :class:`Headers`180 object, a :class:`MultiDict`, :class:`dict`, or iterable of181 pairs.182 183 .. versionchanged:: 1.0184 Support :class:`MultiDict`. Allow passing ``kwargs``.185 """186 if len(args) > 1:187 raise TypeError(f"update expected at most 1 arguments, got {len(args)}")188 189 if args:190 for key, value in iter_multi_items(args[0]):191 self.add(key, value)192 193 for key, value in iter_multi_items(kwargs):194 self.add(key, value)195 196 def __delitem__(self, key, _index_operation=True):197 if _index_operation and isinstance(key, (int, slice)):198 del self._list[key]199 return200 key = key.lower()201 new = []202 for k, v in self._list:203 if k.lower() != key:204 new.append((k, v))205 self._list[:] = new206 207 def remove(self, key):208 """Remove a key.209 210 :param key: The key to be removed.211 """212 return self.__delitem__(key, _index_operation=False)213 214 def pop(self, key=None, default=_missing):215 """Removes and returns a key or index.216 217 :param key: The key to be popped. If this is an integer the item at218 that position is removed, if it's a string the value for219 that key is. If the key is omitted or `None` the last220 item is removed.221 :return: an item.222 """223 if key is None:224 return self._list.pop()225 if isinstance(key, int):226 return self._list.pop(key)227 try:228 rv = self[key]229 self.remove(key)230 except KeyError:231 if default is not _missing:232 return default233 raise234 return rv235 236 def popitem(self):237 """Removes a key or index and returns a (key, value) item."""238 return self.pop()239 240 def __contains__(self, key):241 """Check if a key is present."""242 try:243 self.__getitem__(key, _get_mode=True)244 except KeyError:245 return False246 return True247 248 def __iter__(self):249 """Yield ``(key, value)`` tuples."""250 return iter(self._list)251 252 def __len__(self):253 return len(self._list)254 255 def add(self, _key, _value, **kw):256 """Add a new header tuple to the list.257 258 Keyword arguments can specify additional parameters for the header259 value, with underscores converted to dashes::260 261 >>> d = Headers()262 >>> d.add('Content-Type', 'text/plain')263 >>> d.add('Content-Disposition', 'attachment', filename='foo.png')264 265 The keyword argument dumping uses :func:`dump_options_header`266 behind the scenes.267 268 .. versionadded:: 0.4.1269 keyword arguments were added for :mod:`wsgiref` compatibility.270 """271 if kw:272 _value = _options_header_vkw(_value, kw)273 _value = _str_header_value(_value)274 self._list.append((_key, _value))275 276 def add_header(self, _key, _value, **_kw):277 """Add a new header tuple to the list.278 279 An alias for :meth:`add` for compatibility with the :mod:`wsgiref`280 :meth:`~wsgiref.headers.Headers.add_header` method.281 """282 self.add(_key, _value, **_kw)283 284 def clear(self):285 """Clears all headers."""286 del self._list[:]287 288 def set(self, _key, _value, **kw):289 """Remove all header tuples for `key` and add a new one. The newly290 added key either appears at the end of the list if there was no291 entry or replaces the first one.292 293 Keyword arguments can specify additional parameters for the header294 value, with underscores converted to dashes. See :meth:`add` for295 more information.296 297 .. versionchanged:: 0.6.1298 :meth:`set` now accepts the same arguments as :meth:`add`.299 300 :param key: The key to be inserted.301 :param value: The value to be inserted.302 """303 if kw:304 _value = _options_header_vkw(_value, kw)305 _value = _str_header_value(_value)306 if not self._list:307 self._list.append((_key, _value))308 return309 listiter = iter(self._list)310 ikey = _key.lower()311 for idx, (old_key, _old_value) in enumerate(listiter):312 if old_key.lower() == ikey:313 # replace first occurrence314 self._list[idx] = (_key, _value)315 break316 else:317 self._list.append((_key, _value))318 return319 self._list[idx + 1 :] = [t for t in listiter if t[0].lower() != ikey]320 321 def setlist(self, key, values):322 """Remove any existing values for a header and add new ones.323 324 :param key: The header key to set.325 :param values: An iterable of values to set for the key.326 327 .. versionadded:: 1.0328 """329 if values:330 values_iter = iter(values)331 self.set(key, next(values_iter))332 333 for value in values_iter:334 self.add(key, value)335 else:336 self.remove(key)337 338 def setdefault(self, key, default):339 """Return the first value for the key if it is in the headers,340 otherwise set the header to the value given by ``default`` and341 return that.342 343 :param key: The header key to get.344 :param default: The value to set for the key if it is not in the345 headers.346 """347 if key in self:348 return self[key]349 350 self.set(key, default)351 return default352 353 def setlistdefault(self, key, default):354 """Return the list of values for the key if it is in the355 headers, otherwise set the header to the list of values given356 by ``default`` and return that.357 358 Unlike :meth:`MultiDict.setlistdefault`, modifying the returned359 list will not affect the headers.360 361 :param key: The header key to get.362 :param default: An iterable of values to set for the key if it363 is not in the headers.364 365 .. versionadded:: 1.0366 """367 if key not in self:368 self.setlist(key, default)369 370 return self.getlist(key)371 372 def __setitem__(self, key, value):373 """Like :meth:`set` but also supports index/slice based setting."""374 if isinstance(key, (slice, int)):375 if isinstance(key, int):376 value = [value]377 value = [(k, _str_header_value(v)) for (k, v) in value]378 if isinstance(key, int):379 self._list[key] = value[0]380 else:381 self._list[key] = value382 else:383 self.set(key, value)384 385 def update(self, *args, **kwargs):386 """Replace headers in this object with items from another387 headers object and keyword arguments.388 389 To extend existing keys instead of replacing, use :meth:`extend`390 instead.391 392 If provided, the first argument can be another :class:`Headers`393 object, a :class:`MultiDict`, :class:`dict`, or iterable of394 pairs.395 396 .. versionadded:: 1.0397 """398 if len(args) > 1:399 raise TypeError(f"update expected at most 1 arguments, got {len(args)}")400 401 if args:402 mapping = args[0]403 404 if isinstance(mapping, (Headers, MultiDict)):405 for key in mapping.keys():406 self.setlist(key, mapping.getlist(key))407 elif isinstance(mapping, dict):408 for key, value in mapping.items():409 if isinstance(value, (list, tuple)):410 self.setlist(key, value)411 else:412 self.set(key, value)413 else:414 for key, value in mapping:415 self.set(key, value)416 417 for key, value in kwargs.items():418 if isinstance(value, (list, tuple)):419 self.setlist(key, value)420 else:421 self.set(key, value)422 423 def to_wsgi_list(self):424 """Convert the headers into a list suitable for WSGI.425 426 :return: list427 """428 return list(self)429 430 def copy(self):431 return self.__class__(self._list)432 433 def __copy__(self):434 return self.copy()435 436 def __str__(self):437 """Returns formatted headers suitable for HTTP transmission."""438 strs = []439 for key, value in self.to_wsgi_list():440 strs.append(f"{key}: {value}")441 strs.append("\r\n")442 return "\r\n".join(strs)443 444 def __repr__(self):445 return f"{type(self).__name__}({list(self)!r})"446 447 448def _options_header_vkw(value: str, kw: dict[str, t.Any]):449 return http.dump_options_header(450 value, {k.replace("_", "-"): v for k, v in kw.items()}451 )452 453 454_newline_re = re.compile(r"[\r\n]")455 456 457def _str_header_value(value: t.Any) -> str:458 if not isinstance(value, str):459 value = str(value)460 461 if _newline_re.search(value) is not None:462 raise ValueError("Header values must not contain newline characters.")463 464 return value465 466 467class EnvironHeaders(ImmutableHeadersMixin, Headers):468 """Read only version of the headers from a WSGI environment. This469 provides the same interface as `Headers` and is constructed from470 a WSGI environment.471 From Werkzeug 0.3 onwards, the `KeyError` raised by this class is also a472 subclass of the :exc:`~exceptions.BadRequest` HTTP exception and will473 render a page for a ``400 BAD REQUEST`` if caught in a catch-all for474 HTTP exceptions.475 """476 477 def __init__(self, environ):478 self.environ = environ479 480 def __eq__(self, other):481 return self.environ is other.environ482 483 __hash__ = None484 485 def __getitem__(self, key, _get_mode=False):486 # _get_mode is a no-op for this class as there is no index but487 # used because get() calls it.488 if not isinstance(key, str):489 raise KeyError(key)490 key = key.upper().replace("-", "_")491 if key in {"CONTENT_TYPE", "CONTENT_LENGTH"}:492 return self.environ[key]493 return self.environ[f"HTTP_{key}"]494 495 def __len__(self):496 # the iter is necessary because otherwise list calls our497 # len which would call list again and so forth.498 return len(list(iter(self)))499 500 def __iter__(self):501 for key, value in self.environ.items():502 if key.startswith("HTTP_") and key not in {503 "HTTP_CONTENT_TYPE",504 "HTTP_CONTENT_LENGTH",505 }:506 yield key[5:].replace("_", "-").title(), value507 elif key in {"CONTENT_TYPE", "CONTENT_LENGTH"} and value:508 yield key.replace("_", "-").title(), value509 510 def copy(self):511 raise TypeError(f"cannot create {type(self).__name__!r} copies")512 513 514# circular dependencies515from .. import http516 