Coverage for /pythoncovmergedfiles/medio/medio/src/aiohttp/aiohttp/helpers.py: 37%
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
1"""Various helper functions"""
3import asyncio
4import base64
5import contextlib
6import dataclasses
7import datetime
8import enum
9import functools
10import inspect
11import netrc
12import os
13import platform
14import re
15import sys
16import time
17import warnings
18import weakref
19from collections.abc import Callable, Iterable, Iterator, Mapping
20from contextlib import suppress
21from email.message import EmailMessage
22from email.parser import HeaderParser
23from email.policy import HTTP
24from email.utils import parsedate
25from http.cookies import SimpleCookie
26from math import ceil
27from pathlib import Path
28from types import MappingProxyType, TracebackType
29from typing import (
30 TYPE_CHECKING,
31 Any,
32 ContextManager,
33 Generic,
34 Protocol,
35 TypeVar,
36 final,
37 get_args,
38 overload,
39)
40from urllib.parse import quote
41from urllib.request import getproxies, proxy_bypass
43from multidict import CIMultiDict, MultiDict, MultiDictProxy
44from propcache.api import under_cached_property as reify
45from yarl import URL
47from . import hdrs
48from .log import client_logger
49from .typedefs import PathLike # noqa
51if TYPE_CHECKING:
52 from dataclasses import dataclass as frozen_dataclass_decorator
53else:
54 frozen_dataclass_decorator = functools.partial(
55 dataclasses.dataclass, frozen=True, slots=True
56 )
58__all__ = ("ChainMapProxy", "ETag", "frozen_dataclass_decorator", "reify")
60# This is the default size/limit for several operations.
61# Matches the max size we receive from sockets:
62# https://github.com/python/cpython/blob/1857a40807daeae3a1bf5efb682de9c9ae6df845/Lib/asyncio/selector_events.py#L766
63DEFAULT_CHUNK_SIZE = 2**18 # 256 KiB
64COOKIE_MAX_LENGTH = 4096
65_QUOTED_PAIR_SUB = re.compile(r"\\(.)")
66_QUOTED_STRING_CONTENT = r'(?:[^"\\]++|\\.)*+'
67_ESCAPED_COMMENT = r"(?:[^()\\]++|\\.)*+"
68_LIST_ELEMENT = rf"""
69[ \t]*
70(?:
71 "( {_QUOTED_STRING_CONTENT} )" # group 1: top-level quoted-string
72 [ \t]* (?:,|\Z)
73 | ( # group 2: unquoted element
74 (?:
75 (?<=[^\s]=) "{_QUOTED_STRING_CONTENT}" # parameter quoted value
76 | (?<=\s) \( {_ESCAPED_COMMENT} \) # comment
77 | [^,"(\\]++ # run of ordinary characters
78 | [^,] # quote, paren or backslash the branches above rejected
79 )++
80 )
81 (?:,|\Z)
82)
83"""
84# Matches one element in a comma-separated header list.
85# Group 1: content of a top-level quoted-string (quotes stripped).
86# Group 2: an unquoted element (may contain parameter quoted-strings / comments).
87_LIST_ELEMENT_RE = re.compile(_LIST_ELEMENT, re.VERBOSE)
88# Finds parameter quoted-strings and comments inside an unquoted element for unescaping.
89_PROTECTED_RE = re.compile(
90 rf"""
91 (?<=[^\s]=) "{_QUOTED_STRING_CONTENT}" # parameter quoted-string
92 | (?<=\s) \( {_ESCAPED_COMMENT} \) # comment
93 """,
94 re.VERBOSE,
95)
97_T = TypeVar("_T")
98_S = TypeVar("_S")
100_SENTINEL = enum.Enum("_SENTINEL", "sentinel")
101sentinel = _SENTINEL.sentinel
103NO_EXTENSIONS = bool(os.environ.get("AIOHTTP_NO_EXTENSIONS"))
105# https://datatracker.ietf.org/doc/html/rfc9112#section-6.3-2.1
106EMPTY_BODY_STATUS_CODES = frozenset((204, 304, *range(100, 200)))
107# https://datatracker.ietf.org/doc/html/rfc9112#section-6.3-2.1
108# https://datatracker.ietf.org/doc/html/rfc9112#section-6.3-2.2
109EMPTY_BODY_METHODS = frozenset({hdrs.METH_HEAD})
111DEBUG = sys.flags.dev_mode or (
112 not sys.flags.ignore_environment and bool(os.environ.get("PYTHONASYNCIODEBUG"))
113)
116EMPTY_SCHEMA_SET = frozenset({""})
117HTTP_SCHEMA_SET = frozenset({"http", "https"})
118WS_SCHEMA_SET = frozenset({"ws", "wss"})
119HTTP_AND_EMPTY_SCHEMA_SET = HTTP_SCHEMA_SET | EMPTY_SCHEMA_SET
120HIGH_LEVEL_SCHEMA_SET = HTTP_AND_EMPTY_SCHEMA_SET | WS_SCHEMA_SET
123CHAR = {chr(i) for i in range(0, 128)}
124CTL = {chr(i) for i in range(0, 32)} | {
125 chr(127),
126}
127SEPARATORS = {
128 "(",
129 ")",
130 "<",
131 ">",
132 "@",
133 ",",
134 ";",
135 ":",
136 "\\",
137 '"',
138 "/",
139 "[",
140 "]",
141 "?",
142 "=",
143 "{",
144 "}",
145 " ",
146 chr(9),
147}
148TOKEN = CHAR ^ CTL ^ SEPARATORS
151json_re = re.compile(r"^(?:application/|[\w.-]+/[\w.+-]+?\+)json$", re.IGNORECASE)
154def encode_basic_auth(login: str, password: str = "", encoding: str = "utf-8") -> str:
155 """Encode HTTP Basic Authentication credentials as an Authorization header value.
157 Returns a string of the form ``"Basic <base64>"`` suitable for use as the
158 value of the ``Authorization`` (or ``Proxy-Authorization``) header.
159 """
160 if ":" in login:
161 raise ValueError('A ":" is not allowed in login (RFC 7617#section-2)')
162 creds = f"{login}:{password}".encode(encoding)
163 return "Basic " + base64.b64encode(creds).decode(encoding)
166def strip_auth_from_url(url: URL) -> tuple[URL, str | None]:
167 """Strip user/password from a URL and return the Authorization header value.
169 Returns a tuple of ``(url_without_credentials, authorization_header_value)``.
170 The header value is ``None`` if no credentials were present.
171 """
172 # Check raw_user and raw_password first as yarl is likely
173 # to already have these values parsed from the netloc in the cache.
174 if url.raw_user is None and url.raw_password is None:
175 return url, None
176 return url.with_user(None), encode_basic_auth(url.user or "", url.password or "")
179def netrc_from_env() -> netrc.netrc | None:
180 """Load netrc from file.
182 Attempt to load it from the path specified by the env-var
183 NETRC or in the default location in the user's home directory.
185 Returns None if it couldn't be found or fails to parse.
186 """
187 netrc_env = os.environ.get("NETRC")
189 if netrc_env is not None:
190 netrc_path = Path(netrc_env)
191 else:
192 try:
193 home_dir = Path.home()
194 except RuntimeError as e:
195 # if pathlib can't resolve home, it may raise a RuntimeError
196 client_logger.debug(
197 "Could not resolve home directory when "
198 "trying to look for .netrc file: %s",
199 e,
200 )
201 return None
203 netrc_path = home_dir / (
204 "_netrc" if platform.system() == "Windows" else ".netrc"
205 )
207 try:
208 return netrc.netrc(str(netrc_path))
209 except netrc.NetrcParseError as e:
210 client_logger.warning("Could not parse .netrc file: %s", e)
211 except OSError as e:
212 netrc_exists = False
213 with contextlib.suppress(OSError):
214 netrc_exists = netrc_path.is_file()
215 # we couldn't read the file (doesn't exist, permissions, etc.)
216 if netrc_env or netrc_exists:
217 # only warn if the environment wanted us to load it,
218 # or it appears like the default file does actually exist
219 client_logger.warning("Could not read .netrc file: %s", e)
221 return None
224@frozen_dataclass_decorator
225class ProxyInfo:
226 proxy: URL
227 proxy_auth: str | None
230def _auth_header_from_netrc(netrc_obj: netrc.netrc | None, host: str) -> str:
231 """Return a ``Proxy-Authorization`` header value for ``host`` from netrc.
233 :raises LookupError: if ``netrc_obj`` is :py:data:`None` or if no
234 entry is found for the ``host``.
235 """
236 if netrc_obj is None:
237 raise LookupError("No .netrc file found")
238 auth_from_netrc = netrc_obj.authenticators(host)
240 if auth_from_netrc is None:
241 raise LookupError(f"No entry for {host!s} found in the `.netrc` file.")
242 login, account, password = auth_from_netrc
244 return encode_basic_auth(login or account, password)
247def proxies_from_env() -> dict[str, ProxyInfo]:
248 proxy_urls = {
249 k: URL(v)
250 for k, v in getproxies().items()
251 if k in ("http", "https", "ws", "wss")
252 }
253 netrc_obj = netrc_from_env()
254 stripped = {k: strip_auth_from_url(v) for k, v in proxy_urls.items()}
255 ret = {}
256 for proto, val in stripped.items():
257 proxy, auth = val
258 if proxy.scheme in ("https", "wss"):
259 client_logger.warning(
260 "%s proxies %s are not supported, ignoring", proxy.scheme.upper(), proxy
261 )
262 continue
263 if netrc_obj and auth is None:
264 if proxy.host is not None:
265 try:
266 auth = _auth_header_from_netrc(netrc_obj, proxy.host)
267 except LookupError:
268 auth = None
269 ret[proto] = ProxyInfo(proxy, auth)
270 return ret
273def get_env_proxy_for_url(url: URL) -> tuple[URL, str | None]:
274 """Get a permitted proxy for the given URL from the env."""
275 if url.host is not None and proxy_bypass(url.host):
276 raise LookupError(f"Proxying is disallowed for `{url.host!r}`")
278 proxies_in_env = proxies_from_env()
279 try:
280 proxy_info = proxies_in_env[url.scheme]
281 except KeyError:
282 raise LookupError(f"No proxies found for `{url!s}` in the env")
283 else:
284 return proxy_info.proxy, proxy_info.proxy_auth
287@frozen_dataclass_decorator
288class MimeType:
289 type: str
290 subtype: str
291 suffix: str
292 parameters: "MultiDictProxy[str]"
295@functools.lru_cache(maxsize=56)
296def parse_mimetype(mimetype: str) -> MimeType:
297 """Parses a MIME type into its components.
299 mimetype is a MIME type string.
301 Returns a MimeType object.
303 Example:
305 >>> parse_mimetype('text/html; charset=utf-8')
306 MimeType(type='text', subtype='html', suffix='',
307 parameters={'charset': 'utf-8'})
309 """
310 if not mimetype:
311 return MimeType(
312 type="", subtype="", suffix="", parameters=MultiDictProxy(MultiDict())
313 )
315 parts = mimetype.split(";")
316 params: MultiDict[str] = MultiDict()
317 for item in parts[1:]:
318 if not item.strip():
319 continue
320 key, _, value = item.partition("=")
321 params.add(key.lower().strip(), value.strip(' "'))
323 fulltype = parts[0].strip().lower()
324 if fulltype == "*":
325 fulltype = "*/*"
327 mtype, _, stype = fulltype.partition("/")
328 stype, _, suffix = stype.partition("+")
330 return MimeType(
331 type=mtype, subtype=stype, suffix=suffix, parameters=MultiDictProxy(params)
332 )
335class EnsureOctetStream(EmailMessage):
336 def __init__(self) -> None:
337 super().__init__()
338 # https://www.rfc-editor.org/rfc/rfc9110#section-8.3-5
339 self.set_default_type("application/octet-stream")
341 def get_content_type(self) -> str:
342 """Re-implementation from Message
344 Returns application/octet-stream in place of plain/text when
345 value is wrong.
347 The way this class is used guarantees that content-type will
348 be present so simplify the checks wrt to the base implementation.
349 """
350 value = self.get("content-type", "").lower()
352 # Based on the implementation of _splitparam in the standard library
353 ctype, _, _ = value.partition(";")
354 ctype = ctype.strip()
355 if ctype.count("/") != 1:
356 return self.get_default_type()
357 return ctype
360@functools.lru_cache(maxsize=56)
361def parse_content_type(raw: str) -> tuple[str, MappingProxyType[str, str]]:
362 """Parse Content-Type header.
364 Returns a tuple of the parsed content type and a
365 MappingProxyType of parameters. The default returned value
366 is `application/octet-stream`
367 """
368 msg = HeaderParser(EnsureOctetStream, policy=HTTP).parsestr(f"Content-Type: {raw}")
369 content_type = msg.get_content_type()
370 params = msg.get_params(())
371 content_dict = dict(params[1:]) # First element is content type again
372 return content_type, MappingProxyType(content_dict)
375def guess_filename(obj: Any, default: str | None = None) -> str | None:
376 name = getattr(obj, "name", None)
377 if name and isinstance(name, str) and name[0] != "<" and name[-1] != ">":
378 return Path(name).name
379 return default
382not_qtext_re = re.compile(r"[^\041\043-\133\135-\176]")
383QCONTENT = {chr(i) for i in range(0x20, 0x7F)} | {"\t"}
386def quoted_string(content: str) -> str:
387 """Return 7-bit content as quoted-string.
389 Format content into a quoted-string as defined in RFC5322 for
390 Internet Message Format. Notice that this is not the 8-bit HTTP
391 format, but the 7-bit email format. Content must be in usascii or
392 a ValueError is raised.
393 """
394 if not (QCONTENT > set(content)):
395 raise ValueError(f"bad content for quoted-string {content!r}")
396 return not_qtext_re.sub(lambda x: "\\" + x.group(0), content)
399def content_disposition_header(
400 disptype: str,
401 quote_fields: bool = True,
402 _charset: str = "utf-8",
403 params: dict[str, str] | None = None,
404) -> str:
405 """Sets ``Content-Disposition`` header for MIME.
407 This is the MIME payload Content-Disposition header from RFC 2183
408 and RFC 7579 section 4.2, not the HTTP Content-Disposition from
409 RFC 6266.
411 disptype is a disposition type: inline, attachment, form-data.
412 Should be valid extension token (see RFC 2183)
414 quote_fields performs value quoting to 7-bit MIME headers
415 according to RFC 7578. Set to quote_fields to False if recipient
416 can take 8-bit file names and field values.
418 _charset specifies the charset to use when quote_fields is True.
420 params is a dict with disposition params.
421 """
422 if not disptype or not (TOKEN > set(disptype)):
423 raise ValueError(f"bad content disposition type {disptype!r}")
425 value = disptype
426 if params:
427 lparams = []
428 for key, val in params.items():
429 if not key or not (TOKEN > set(key)):
430 raise ValueError(f"bad content disposition parameter {key!r}={val!r}")
431 if quote_fields:
432 if key.lower() == "filename":
433 qval = quote(val, "", encoding=_charset)
434 lparams.append((key, '"%s"' % qval))
435 else:
436 try:
437 qval = quoted_string(val)
438 except ValueError:
439 qval = "".join(
440 (_charset, "''", quote(val, "", encoding=_charset))
441 )
442 lparams.append((key + "*", qval))
443 else:
444 lparams.append((key, '"%s"' % qval))
445 else:
446 qval = val.replace("\\", "\\\\").replace('"', '\\"')
447 lparams.append((key, '"%s"' % qval))
448 sparams = "; ".join("=".join(pair) for pair in lparams)
449 value = "; ".join((value, sparams))
450 return value
453def is_expected_content_type(
454 response_content_type: str, expected_content_type: str
455) -> bool:
456 """Checks if received content type is processable as an expected one.
458 Both arguments should be given without parameters.
459 """
460 if expected_content_type == "application/json":
461 return json_re.match(response_content_type) is not None
462 return expected_content_type in response_content_type
465def is_ip_address(host: str | None) -> bool:
466 """Check if host looks like an IP Address.
468 This check is only meant as a heuristic to ensure that
469 a host is not a domain name.
470 """
471 if not host:
472 return False
473 # For a host to be an ipv4 address, it must be all numeric.
474 # The host must contain a colon to be an IPv6 address.
475 return ":" in host or host.replace(".", "").isdigit()
478def is_canonical_ipv4_address(host: str) -> bool:
479 """Check if host is a canonical dotted-quad IPv4 address.
481 Rejects the legacy numeric forms that ``socket`` still accepts and
482 maps onto an address, e.g. ``2130706433``, ``017700000001``, ``127.1``.
483 """
484 parts = host.split(".")
485 if len(parts) != 4:
486 return False
487 for part in parts:
488 # Each octet must be 1-3 ASCII digits; reject unicode digits
489 # (which ``str.isdigit`` accepts but ``int`` may not), octal
490 # leading zeros, and values above 255.
491 if not (1 <= len(part) <= 3) or not part.isascii() or not part.isdigit():
492 return False
493 if part[0] == "0" and len(part) != 1:
494 return False
495 if int(part) > 255:
496 return False
497 return True
500_cached_current_datetime: int | None = None
501_cached_formatted_datetime = ""
504def rfc822_formatted_time() -> str:
505 global _cached_current_datetime
506 global _cached_formatted_datetime
508 now = int(time.time())
509 if now != _cached_current_datetime:
510 # Weekday and month names for HTTP date/time formatting;
511 # always English!
512 # Tuples are constants stored in codeobject!
513 _weekdayname = ("Mon", "Tue", "Wed", "Thu", "Fri", "Sat", "Sun")
514 _monthname = (
515 "", # Dummy so we can use 1-based month numbers
516 "Jan",
517 "Feb",
518 "Mar",
519 "Apr",
520 "May",
521 "Jun",
522 "Jul",
523 "Aug",
524 "Sep",
525 "Oct",
526 "Nov",
527 "Dec",
528 )
530 year, month, day, hh, mm, ss, wd, *tail = time.gmtime(now)
531 _cached_formatted_datetime = "%s, %02d %3s %4d %02d:%02d:%02d GMT" % (
532 _weekdayname[wd],
533 day,
534 _monthname[month],
535 year,
536 hh,
537 mm,
538 ss,
539 )
540 _cached_current_datetime = now
541 return _cached_formatted_datetime
544def _weakref_handle(info: "tuple[weakref.ref[object], str]") -> None:
545 ref, name = info
546 ob = ref()
547 if ob is not None:
548 with suppress(Exception):
549 getattr(ob, name)()
552def weakref_handle(
553 ob: object,
554 name: str,
555 timeout: float | None,
556 loop: asyncio.AbstractEventLoop,
557 timeout_ceil_threshold: float = 5,
558) -> asyncio.TimerHandle | None:
559 if timeout is not None and timeout > 0:
560 when = loop.time() + timeout
561 if timeout >= timeout_ceil_threshold:
562 when = ceil(when)
564 return loop.call_at(when, _weakref_handle, (weakref.ref(ob), name))
565 return None
568def call_later(
569 cb: Callable[[], Any],
570 timeout: float | None,
571 loop: asyncio.AbstractEventLoop,
572 timeout_ceil_threshold: float = 5,
573) -> asyncio.TimerHandle | None:
574 if timeout is None or timeout <= 0:
575 return None
576 now = loop.time()
577 when = calculate_timeout_when(now, timeout, timeout_ceil_threshold)
578 return loop.call_at(when, cb)
581def calculate_timeout_when(
582 loop_time: float,
583 timeout: float,
584 timeout_ceiling_threshold: float,
585) -> float:
586 """Calculate when to execute a timeout."""
587 when = loop_time + timeout
588 if timeout > timeout_ceiling_threshold:
589 return ceil(when)
590 return when
593class TimeoutHandle:
594 """Timeout handle"""
596 __slots__ = ("_timeout", "_loop", "_ceil_threshold", "_callbacks")
598 def __init__(
599 self,
600 loop: asyncio.AbstractEventLoop,
601 timeout: float | None,
602 ceil_threshold: float = 5,
603 ) -> None:
604 self._timeout = timeout
605 self._loop = loop
606 self._ceil_threshold = ceil_threshold
607 self._callbacks: list[
608 tuple[Callable[..., None], tuple[Any, ...], dict[str, Any]]
609 ] = []
611 def register(
612 self, callback: Callable[..., None], *args: Any, **kwargs: Any
613 ) -> None:
614 self._callbacks.append((callback, args, kwargs))
616 def close(self) -> None:
617 self._callbacks.clear()
619 def start(self) -> asyncio.TimerHandle | None:
620 timeout = self._timeout
621 if timeout is not None and timeout > 0:
622 when = self._loop.time() + timeout
623 if timeout >= self._ceil_threshold:
624 when = ceil(when)
625 return self._loop.call_at(when, self.__call__)
626 else:
627 return None
629 def timer(self) -> "BaseTimerContext":
630 if self._timeout is not None and self._timeout > 0:
631 timer = TimerContext(self._loop)
632 self.register(timer.timeout)
633 return timer
634 else:
635 return TimerNoop()
637 def __call__(self) -> None:
638 for cb, args, kwargs in self._callbacks:
639 with suppress(Exception):
640 cb(*args, **kwargs)
642 self._callbacks.clear()
645class BaseTimerContext(ContextManager["BaseTimerContext"]):
647 __slots__ = ()
649 def assert_timeout(self) -> None:
650 """Raise TimeoutError if timeout has been exceeded."""
653class TimerNoop(BaseTimerContext):
655 __slots__ = ()
657 def __enter__(self) -> BaseTimerContext:
658 return self
660 def __exit__(
661 self,
662 exc_type: type[BaseException] | None,
663 exc_val: BaseException | None,
664 exc_tb: TracebackType | None,
665 ) -> None:
666 return
669class TimerContext(BaseTimerContext):
670 """Low resolution timeout context manager"""
672 __slots__ = ("_loop", "_tasks", "_cancelled", "_cancelling")
674 def __init__(self, loop: asyncio.AbstractEventLoop) -> None:
675 self._loop = loop
676 self._tasks: list[asyncio.Task[Any]] = []
677 self._cancelled = False
678 self._cancelling = 0
680 def assert_timeout(self) -> None:
681 """Raise TimeoutError if timer has already been cancelled."""
682 if self._cancelled:
683 raise TimeoutError from None
685 def __enter__(self) -> BaseTimerContext:
686 task = asyncio.current_task(loop=self._loop)
687 if task is None:
688 raise RuntimeError("Timeout context manager should be used inside a task")
690 # Remember if the task was already cancelling
691 # so when we __exit__ we can decide if we should
692 # raise TimeoutError or let the cancellation propagate
693 self._cancelling = task.cancelling()
695 if self._cancelled:
696 raise TimeoutError from None
698 self._tasks.append(task)
699 return self
701 def __exit__(
702 self,
703 exc_type: type[BaseException] | None,
704 exc_val: BaseException | None,
705 exc_tb: TracebackType | None,
706 ) -> bool | None:
707 enter_task: asyncio.Task[Any] | None = None
708 if self._tasks:
709 enter_task = self._tasks.pop()
711 if exc_type is asyncio.CancelledError and self._cancelled:
712 assert enter_task is not None
713 # The timeout was hit, and the task was cancelled
714 # so we need to uncancel the last task that entered the context manager
715 # since the cancellation should not leak out of the context manager
716 # If the task was already cancelling don't raise
717 # TimeoutError and instead return None
718 # to allow the cancellation to propagate
719 if enter_task.uncancel() > self._cancelling:
720 return None
721 raise TimeoutError from exc_val
722 return None
724 def timeout(self) -> None:
725 if not self._cancelled:
726 for task in set(self._tasks):
727 task.cancel()
729 self._cancelled = True
732def ceil_timeout(delay: float | None, ceil_threshold: float = 5) -> asyncio.Timeout:
733 if delay is None or delay <= 0:
734 return asyncio.timeout(None)
736 loop = asyncio.get_running_loop()
737 now = loop.time()
738 when = now + delay
739 if delay > ceil_threshold:
740 when = ceil(when)
741 return asyncio.timeout_at(when)
744class HeadersDictProxy(Mapping[str, str]):
745 def __init__(self, md: CIMultiDict[str]):
746 self._md = md
748 def getall(self, key: str) -> tuple[str, ...]:
749 val = self.get(key, "")
750 unescape = _QUOTED_PAIR_SUB.sub
751 values = []
752 for m in _LIST_ELEMENT_RE.finditer(val):
753 qs = m.group(1)
754 if qs is not None:
755 values.append(unescape(r"\1", qs))
756 else:
757 raw = m.group(2).strip()
758 if raw:
759 values.append(
760 _PROTECTED_RE.sub(lambda p: unescape(r"\1", p.group()), raw)
761 )
762 return tuple(values)
764 def __eq__(self, other: object) -> bool:
765 return self._md.__eq__(other)
767 def __getitem__(self, key: str) -> str:
768 return ", ".join(self._md.getall(key))
770 def __iter__(self) -> Iterator[str]:
771 # We need to deduplicate keys from MultiDict
772 # But, we also need to retain ordering
773 seen = set()
774 for k in self._md.__iter__():
775 if k in seen:
776 continue
777 seen.add(k)
778 yield k
780 def __len__(self) -> int:
781 return len(set(self._md.keys()))
783 def __repr__(self) -> str:
784 body = ", ".join(f"'{k}': {v!r}" for k, v in self.items())
785 return f"<{self.__class__.__name__}({body})>"
788class HeadersMixin:
789 """Mixin for handling headers."""
791 _headers: Mapping[str, str]
792 _content_type: str | None = None
793 _content_dict: dict[str, str] | None = None
794 _stored_content_type: str | None | _SENTINEL = sentinel
796 def _parse_content_type(self, raw: str | None) -> None:
797 self._stored_content_type = raw
798 if raw is None:
799 # default value according to RFC 2616
800 self._content_type = "application/octet-stream"
801 self._content_dict = {}
802 else:
803 content_type, content_mapping_proxy = parse_content_type(raw)
804 self._content_type = content_type
805 # _content_dict needs to be mutable so we can update it
806 self._content_dict = content_mapping_proxy.copy()
808 @property
809 def content_type(self) -> str:
810 """The value of content part for Content-Type HTTP header."""
811 raw = self._headers.get(hdrs.CONTENT_TYPE)
812 if self._stored_content_type != raw:
813 self._parse_content_type(raw)
814 assert self._content_type is not None
815 return self._content_type
817 @property
818 def charset(self) -> str | None:
819 """The value of charset part for Content-Type HTTP header."""
820 raw = self._headers.get(hdrs.CONTENT_TYPE)
821 if self._stored_content_type != raw:
822 self._parse_content_type(raw)
823 assert self._content_dict is not None
824 return self._content_dict.get("charset")
826 @property
827 def content_length(self) -> int | None:
828 """The value of Content-Length HTTP header."""
829 content_length = self._headers.get(hdrs.CONTENT_LENGTH)
830 return None if content_length is None else int(content_length)
833def set_result(fut: "asyncio.Future[_T]", result: _T) -> None:
834 if not fut.done():
835 fut.set_result(result)
838_EXC_SENTINEL = BaseException()
841class ErrorableProtocol(Protocol):
842 def set_exception(
843 self,
844 exc: type[BaseException] | BaseException,
845 exc_cause: BaseException = ...,
846 ) -> None: ...
849def set_exception(
850 fut: "asyncio.Future[_T] | ErrorableProtocol",
851 exc: type[BaseException] | BaseException,
852 exc_cause: BaseException = _EXC_SENTINEL,
853) -> None:
854 """Set future exception.
856 If the future is marked as complete, this function is a no-op.
858 :param exc_cause: An exception that is a direct cause of ``exc``.
859 Only set if provided.
860 """
861 if asyncio.isfuture(fut) and fut.done():
862 return
864 exc_is_sentinel = exc_cause is _EXC_SENTINEL
865 exc_causes_itself = exc is exc_cause
866 if not exc_is_sentinel and not exc_causes_itself:
867 exc.__cause__ = exc_cause
869 fut.set_exception(exc)
872@functools.total_ordering
873class BaseKey(Generic[_T]):
874 """Base for concrete context storage key classes.
876 Each storage is provided with its own sub-class for the sake of some additional type safety.
877 """
879 __slots__ = ("_name", "_t", "__orig_class__")
881 # This may be set by Python when instantiating with a generic type. We need to
882 # support this, in order to support types that are not concrete classes,
883 # like Iterable, which can't be passed as the second parameter to __init__.
884 __orig_class__: type[object]
886 # TODO(PY314): Change Type to TypeForm (this should resolve unreachable below).
887 def __init__(self, name: str, t: type[_T] | None = None):
888 # Prefix with module name to help deduplicate key names.
889 frame = inspect.currentframe()
890 while frame:
891 if frame.f_code.co_name == "<module>":
892 module: str = frame.f_globals["__name__"]
893 break
894 frame = frame.f_back
895 else:
896 raise RuntimeError("Failed to get module name.")
898 # https://github.com/python/mypy/issues/14209
899 self._name = module + "." + name # type: ignore[possibly-undefined]
900 self._t = t
902 def __lt__(self, other: object) -> bool:
903 if isinstance(other, BaseKey):
904 return self._name < other._name
905 return True # Order BaseKey above other types.
907 def __repr__(self) -> str:
908 t = self._t
909 if t is None:
910 with suppress(AttributeError):
911 # Set to type arg.
912 t = get_args(self.__orig_class__)[0]
914 if t is None:
915 t_repr = "<<Unknown>>"
916 elif isinstance(t, type):
917 if t.__module__ == "builtins":
918 t_repr = t.__qualname__
919 else:
920 t_repr = f"{t.__module__}.{t.__qualname__}"
921 else:
922 t_repr = repr(t) # type: ignore[unreachable]
923 return f"<{self.__class__.__name__}({self._name}, type={t_repr})>"
926class AppKey(BaseKey[_T]):
927 """Keys for static typing support in Application."""
930class RequestKey(BaseKey[_T]):
931 """Keys for static typing support in Request."""
934class ResponseKey(BaseKey[_T]):
935 """Keys for static typing support in Response."""
938@final
939class ChainMapProxy(Mapping[str | AppKey[Any], Any]):
940 __slots__ = ("_maps",)
942 def __init__(self, maps: Iterable[Mapping[str | AppKey[Any], Any]]) -> None:
943 self._maps = tuple(maps)
945 def __init_subclass__(cls) -> None:
946 raise TypeError(
947 f"Inheritance class {cls.__name__} from ChainMapProxy is forbidden"
948 )
950 @overload
951 def __getitem__(self, key: AppKey[_T]) -> _T: ...
953 @overload
954 def __getitem__(self, key: str) -> Any: ...
956 def __getitem__(self, key: str | AppKey[_T]) -> Any:
957 for mapping in self._maps:
958 try:
959 return mapping[key]
960 except KeyError:
961 pass
962 raise KeyError(key)
964 @overload # type: ignore[override]
965 def get(self, key: AppKey[_T], default: _S) -> _T | _S: ...
967 @overload
968 def get(self, key: AppKey[_T], default: None = ...) -> _T | None: ...
970 @overload
971 def get(self, key: str, default: Any = ...) -> Any: ...
973 def get(self, key: str | AppKey[_T], default: Any = None) -> Any:
974 try:
975 return self[key]
976 except KeyError:
977 return default
979 def __len__(self) -> int:
980 # reuses stored hash values if possible
981 return len(set().union(*self._maps))
983 def __iter__(self) -> Iterator[str | AppKey[Any]]:
984 d: dict[str | AppKey[Any], Any] = {}
985 for mapping in reversed(self._maps):
986 # reuses stored hash values if possible
987 d.update(mapping)
988 return iter(d)
990 def __contains__(self, key: object) -> bool:
991 return any(key in m for m in self._maps)
993 def __bool__(self) -> bool:
994 return any(self._maps)
996 def __repr__(self) -> str:
997 content = ", ".join(map(repr, self._maps))
998 return f"ChainMapProxy({content})"
1001class CookieMixin:
1002 """Mixin for handling cookies."""
1004 _cookies: SimpleCookie | None = None
1006 @property
1007 def cookies(self) -> SimpleCookie:
1008 if self._cookies is None:
1009 self._cookies = SimpleCookie()
1010 return self._cookies
1012 def set_cookie(
1013 self,
1014 name: str,
1015 value: str,
1016 *,
1017 expires: str | None = None,
1018 domain: str | None = None,
1019 max_age: int | str | None = None,
1020 path: str = "/",
1021 secure: bool | None = None,
1022 httponly: bool | None = None,
1023 samesite: str | None = None,
1024 partitioned: bool | None = None,
1025 ) -> None:
1026 """Set or update response cookie.
1028 Sets new cookie or updates existent with new value.
1029 Also updates only those params which are not None.
1030 """
1031 if self._cookies is None:
1032 self._cookies = SimpleCookie()
1034 self._cookies[name] = value
1035 c = self._cookies[name]
1037 if expires is not None:
1038 c["expires"] = expires
1039 elif c.get("expires") == "Thu, 01 Jan 1970 00:00:00 GMT":
1040 del c["expires"]
1042 if domain is not None:
1043 c["domain"] = domain
1045 if max_age is not None:
1046 c["max-age"] = str(max_age)
1047 elif "max-age" in c:
1048 del c["max-age"]
1050 c["path"] = path
1052 if secure is not None:
1053 c["secure"] = secure
1054 if httponly is not None:
1055 c["httponly"] = httponly
1056 if samesite is not None:
1057 c["samesite"] = samesite
1059 if partitioned is not None:
1060 c["partitioned"] = partitioned
1062 if DEBUG:
1063 cookie_length = len(c.output(header="")[1:])
1064 if cookie_length > COOKIE_MAX_LENGTH:
1065 warnings.warn(
1066 "The size of is too large, it might get ignored by the client.",
1067 UserWarning,
1068 stacklevel=2,
1069 )
1071 def del_cookie(
1072 self,
1073 name: str,
1074 *,
1075 domain: str | None = None,
1076 path: str = "/",
1077 secure: bool | None = None,
1078 httponly: bool | None = None,
1079 samesite: str | None = None,
1080 ) -> None:
1081 """Delete cookie.
1083 Creates new empty expired cookie.
1084 """
1085 # TODO: do we need domain/path here?
1086 if self._cookies is not None:
1087 self._cookies.pop(name, None)
1088 self.set_cookie(
1089 name,
1090 "",
1091 max_age=0,
1092 expires="Thu, 01 Jan 1970 00:00:00 GMT",
1093 domain=domain,
1094 path=path,
1095 secure=secure,
1096 httponly=httponly,
1097 samesite=samesite,
1098 )
1101def populate_with_cookies(headers: "CIMultiDict[str]", cookies: SimpleCookie) -> None:
1102 for cookie in cookies.values():
1103 value = cookie.output(header="")[1:]
1104 headers.add(hdrs.SET_COOKIE, value)
1107# https://tools.ietf.org/html/rfc7232#section-2.3
1108_ETAGC = r"[!\x23-\x7E\x80-\xff]+"
1109_ETAGC_RE = re.compile(_ETAGC)
1110_QUOTED_ETAG = rf'(W/)?"({_ETAGC})"'
1111QUOTED_ETAG_RE = re.compile(_QUOTED_ETAG)
1112LIST_QUOTED_ETAG_RE = re.compile(rf"({_QUOTED_ETAG})(?:\s*,\s*|$)|(.)")
1114ETAG_ANY = "*"
1117@frozen_dataclass_decorator
1118class ETag:
1119 value: str
1120 is_weak: bool = False
1123def validate_etag_value(value: str) -> None:
1124 if value != ETAG_ANY and not _ETAGC_RE.fullmatch(value):
1125 raise ValueError(
1126 f"Value {value!r} is not a valid etag. Maybe it contains '\"'?"
1127 )
1130def parse_http_date(date_str: str | None) -> datetime.datetime | None:
1131 """Process a date string, return a datetime object"""
1132 if date_str is not None:
1133 timetuple = parsedate(date_str)
1134 if timetuple is not None:
1135 with suppress(ValueError):
1136 return datetime.datetime(*timetuple[:6], tzinfo=datetime.UTC)
1137 return None
1140@functools.lru_cache
1141def must_be_empty_body(method: str, code: int) -> bool:
1142 """Check if a request must return an empty body."""
1143 return (
1144 code in EMPTY_BODY_STATUS_CODES
1145 or method in EMPTY_BODY_METHODS
1146 or (200 <= code < 300 and method == hdrs.METH_CONNECT)
1147 )
1150def should_remove_content_length(method: str, code: int) -> bool:
1151 """Check if a Content-Length header should be removed.
1153 This should always be a subset of must_be_empty_body
1154 """
1155 # https://www.rfc-editor.org/rfc/rfc9110.html#section-8.6-8
1156 # https://www.rfc-editor.org/rfc/rfc9110.html#section-15.4.5-4
1157 return code in EMPTY_BODY_STATUS_CODES or (
1158 200 <= code < 300 and method == hdrs.METH_CONNECT
1159 )