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 Union,
37 final,
38 get_args,
39 overload,
40)
41from urllib.parse import quote
42from urllib.request import getproxies, proxy_bypass
44from multidict import CIMultiDict, MultiDict, MultiDictProxy
45from propcache.api import under_cached_property as reify
46from yarl import URL
48from . import hdrs
49from .log import client_logger
50from .typedefs import PathLike # noqa
52if sys.version_info >= (3, 11):
53 import asyncio as async_timeout
54else:
55 import async_timeout
57if TYPE_CHECKING:
58 from dataclasses import dataclass as frozen_dataclass_decorator
59else:
60 frozen_dataclass_decorator = functools.partial(
61 dataclasses.dataclass, frozen=True, slots=True
62 )
64__all__ = ("ChainMapProxy", "ETag", "frozen_dataclass_decorator", "reify")
66# This is the default size/limit for several operations.
67# Matches the max size we receive from sockets:
68# https://github.com/python/cpython/blob/1857a40807daeae3a1bf5efb682de9c9ae6df845/Lib/asyncio/selector_events.py#L766
69DEFAULT_CHUNK_SIZE = 2**18 # 256 KiB
70COOKIE_MAX_LENGTH = 4096
71_QUOTED_PAIR_SUB = re.compile(r"\\(.)")
72_QUOTED_STRING = r'"(?:[^"\\]|\\.)*"'
73_ESCAPED_COMMENT = r"(?:[^()\\]|\\.)*"
74# Matches one element in a comma-separated header list.
75# Group 1: content of a top-level quoted-string (quotes stripped).
76# Group 2: an unquoted element (may contain parameter quoted-strings / comments).
77_LIST_ELEMENT_RE = re.compile(
78 rf"""
79 [ \t]*
80 (?:
81 "( (?:[^"\\]|\\.)* )" # group 1: top-level quoted-string
82 | ( # group 2: unquoted element
83 (?:
84 (?<=[^\s]=) {_QUOTED_STRING} # parameter quoted value
85 | (?<=\s) \( {_ESCAPED_COMMENT} \) # comment
86 | [^,] # any non-comma character
87 )+?
88 )
89 )
90 [ \t]* (?:,|\Z)
91 """,
92 re.VERBOSE,
93)
94# Finds parameter quoted-strings and comments inside an unquoted element for unescaping.
95_PROTECTED_RE = re.compile(
96 rf"""
97 (?<=[^\s]=) {_QUOTED_STRING} # parameter quoted-string
98 | (?<=\s) \( {_ESCAPED_COMMENT} \) # comment
99 """,
100 re.VERBOSE,
101)
103_T = TypeVar("_T")
104_S = TypeVar("_S")
106_SENTINEL = enum.Enum("_SENTINEL", "sentinel")
107sentinel = _SENTINEL.sentinel
109NO_EXTENSIONS = bool(os.environ.get("AIOHTTP_NO_EXTENSIONS"))
111# https://datatracker.ietf.org/doc/html/rfc9112#section-6.3-2.1
112EMPTY_BODY_STATUS_CODES = frozenset((204, 304, *range(100, 200)))
113# https://datatracker.ietf.org/doc/html/rfc9112#section-6.3-2.1
114# https://datatracker.ietf.org/doc/html/rfc9112#section-6.3-2.2
115EMPTY_BODY_METHODS = frozenset({hdrs.METH_HEAD})
117DEBUG = sys.flags.dev_mode or (
118 not sys.flags.ignore_environment and bool(os.environ.get("PYTHONASYNCIODEBUG"))
119)
122EMPTY_SCHEMA_SET = frozenset({""})
123HTTP_SCHEMA_SET = frozenset({"http", "https"})
124WS_SCHEMA_SET = frozenset({"ws", "wss"})
125HTTP_AND_EMPTY_SCHEMA_SET = HTTP_SCHEMA_SET | EMPTY_SCHEMA_SET
126HIGH_LEVEL_SCHEMA_SET = HTTP_AND_EMPTY_SCHEMA_SET | WS_SCHEMA_SET
129CHAR = {chr(i) for i in range(0, 128)}
130CTL = {chr(i) for i in range(0, 32)} | {
131 chr(127),
132}
133SEPARATORS = {
134 "(",
135 ")",
136 "<",
137 ">",
138 "@",
139 ",",
140 ";",
141 ":",
142 "\\",
143 '"',
144 "/",
145 "[",
146 "]",
147 "?",
148 "=",
149 "{",
150 "}",
151 " ",
152 chr(9),
153}
154TOKEN = CHAR ^ CTL ^ SEPARATORS
157json_re = re.compile(r"^(?:application/|[\w.-]+/[\w.+-]+?\+)json$", re.IGNORECASE)
160def encode_basic_auth(login: str, password: str = "", encoding: str = "utf-8") -> str:
161 """Encode HTTP Basic Authentication credentials as an Authorization header value.
163 Returns a string of the form ``"Basic <base64>"`` suitable for use as the
164 value of the ``Authorization`` (or ``Proxy-Authorization``) header.
165 """
166 if ":" in login:
167 raise ValueError('A ":" is not allowed in login (RFC 7617#section-2)')
168 creds = f"{login}:{password}".encode(encoding)
169 return "Basic " + base64.b64encode(creds).decode(encoding)
172def strip_auth_from_url(url: URL) -> tuple[URL, str | None]:
173 """Strip user/password from a URL and return the Authorization header value.
175 Returns a tuple of ``(url_without_credentials, authorization_header_value)``.
176 The header value is ``None`` if no credentials were present.
177 """
178 # Check raw_user and raw_password first as yarl is likely
179 # to already have these values parsed from the netloc in the cache.
180 if url.raw_user is None and url.raw_password is None:
181 return url, None
182 return url.with_user(None), encode_basic_auth(url.user or "", url.password or "")
185def netrc_from_env() -> netrc.netrc | None:
186 """Load netrc from file.
188 Attempt to load it from the path specified by the env-var
189 NETRC or in the default location in the user's home directory.
191 Returns None if it couldn't be found or fails to parse.
192 """
193 netrc_env = os.environ.get("NETRC")
195 if netrc_env is not None:
196 netrc_path = Path(netrc_env)
197 else:
198 try:
199 home_dir = Path.home()
200 except RuntimeError as e:
201 # if pathlib can't resolve home, it may raise a RuntimeError
202 client_logger.debug(
203 "Could not resolve home directory when "
204 "trying to look for .netrc file: %s",
205 e,
206 )
207 return None
209 netrc_path = home_dir / (
210 "_netrc" if platform.system() == "Windows" else ".netrc"
211 )
213 try:
214 return netrc.netrc(str(netrc_path))
215 except netrc.NetrcParseError as e:
216 client_logger.warning("Could not parse .netrc file: %s", e)
217 except OSError as e:
218 netrc_exists = False
219 with contextlib.suppress(OSError):
220 netrc_exists = netrc_path.is_file()
221 # we couldn't read the file (doesn't exist, permissions, etc.)
222 if netrc_env or netrc_exists:
223 # only warn if the environment wanted us to load it,
224 # or it appears like the default file does actually exist
225 client_logger.warning("Could not read .netrc file: %s", e)
227 return None
230@frozen_dataclass_decorator
231class ProxyInfo:
232 proxy: URL
233 proxy_auth: str | None
236def _auth_header_from_netrc(netrc_obj: netrc.netrc | None, host: str) -> str:
237 """Return a ``Proxy-Authorization`` header value for ``host`` from netrc.
239 :raises LookupError: if ``netrc_obj`` is :py:data:`None` or if no
240 entry is found for the ``host``.
241 """
242 if netrc_obj is None:
243 raise LookupError("No .netrc file found")
244 auth_from_netrc = netrc_obj.authenticators(host)
246 if auth_from_netrc is None:
247 raise LookupError(f"No entry for {host!s} found in the `.netrc` file.")
248 login, account, password = auth_from_netrc
250 # TODO(PY311): username = login or account
251 # Up to python 3.10, account could be None if not specified,
252 # and login will be empty string if not specified. From 3.11,
253 # login and account will be empty string if not specified.
254 username = login if (login or account is None) else account
256 # TODO(PY311): Remove this, as password will be empty string
257 # if not specified
258 if password is None:
259 password = "" # type: ignore[unreachable]
261 return encode_basic_auth(username, password)
264def proxies_from_env() -> dict[str, ProxyInfo]:
265 proxy_urls = {
266 k: URL(v)
267 for k, v in getproxies().items()
268 if k in ("http", "https", "ws", "wss")
269 }
270 netrc_obj = netrc_from_env()
271 stripped = {k: strip_auth_from_url(v) for k, v in proxy_urls.items()}
272 ret = {}
273 for proto, val in stripped.items():
274 proxy, auth = val
275 if proxy.scheme in ("https", "wss"):
276 client_logger.warning(
277 "%s proxies %s are not supported, ignoring", proxy.scheme.upper(), proxy
278 )
279 continue
280 if netrc_obj and auth is None:
281 if proxy.host is not None:
282 try:
283 auth = _auth_header_from_netrc(netrc_obj, proxy.host)
284 except LookupError:
285 auth = None
286 ret[proto] = ProxyInfo(proxy, auth)
287 return ret
290def get_env_proxy_for_url(url: URL) -> tuple[URL, str | None]:
291 """Get a permitted proxy for the given URL from the env."""
292 if url.host is not None and proxy_bypass(url.host):
293 raise LookupError(f"Proxying is disallowed for `{url.host!r}`")
295 proxies_in_env = proxies_from_env()
296 try:
297 proxy_info = proxies_in_env[url.scheme]
298 except KeyError:
299 raise LookupError(f"No proxies found for `{url!s}` in the env")
300 else:
301 return proxy_info.proxy, proxy_info.proxy_auth
304@frozen_dataclass_decorator
305class MimeType:
306 type: str
307 subtype: str
308 suffix: str
309 parameters: "MultiDictProxy[str]"
312@functools.lru_cache(maxsize=56)
313def parse_mimetype(mimetype: str) -> MimeType:
314 """Parses a MIME type into its components.
316 mimetype is a MIME type string.
318 Returns a MimeType object.
320 Example:
322 >>> parse_mimetype('text/html; charset=utf-8')
323 MimeType(type='text', subtype='html', suffix='',
324 parameters={'charset': 'utf-8'})
326 """
327 if not mimetype:
328 return MimeType(
329 type="", subtype="", suffix="", parameters=MultiDictProxy(MultiDict())
330 )
332 parts = mimetype.split(";")
333 params: MultiDict[str] = MultiDict()
334 for item in parts[1:]:
335 if not item.strip():
336 continue
337 key, _, value = item.partition("=")
338 params.add(key.lower().strip(), value.strip(' "'))
340 fulltype = parts[0].strip().lower()
341 if fulltype == "*":
342 fulltype = "*/*"
344 mtype, _, stype = fulltype.partition("/")
345 stype, _, suffix = stype.partition("+")
347 return MimeType(
348 type=mtype, subtype=stype, suffix=suffix, parameters=MultiDictProxy(params)
349 )
352class EnsureOctetStream(EmailMessage):
353 def __init__(self) -> None:
354 super().__init__()
355 # https://www.rfc-editor.org/rfc/rfc9110#section-8.3-5
356 self.set_default_type("application/octet-stream")
358 def get_content_type(self) -> str:
359 """Re-implementation from Message
361 Returns application/octet-stream in place of plain/text when
362 value is wrong.
364 The way this class is used guarantees that content-type will
365 be present so simplify the checks wrt to the base implementation.
366 """
367 value = self.get("content-type", "").lower()
369 # Based on the implementation of _splitparam in the standard library
370 ctype, _, _ = value.partition(";")
371 ctype = ctype.strip()
372 if ctype.count("/") != 1:
373 return self.get_default_type()
374 return ctype
377@functools.lru_cache(maxsize=56)
378def parse_content_type(raw: str) -> tuple[str, MappingProxyType[str, str]]:
379 """Parse Content-Type header.
381 Returns a tuple of the parsed content type and a
382 MappingProxyType of parameters. The default returned value
383 is `application/octet-stream`
384 """
385 msg = HeaderParser(EnsureOctetStream, policy=HTTP).parsestr(f"Content-Type: {raw}")
386 content_type = msg.get_content_type()
387 params = msg.get_params(())
388 content_dict = dict(params[1:]) # First element is content type again
389 return content_type, MappingProxyType(content_dict)
392def guess_filename(obj: Any, default: str | None = None) -> str | None:
393 name = getattr(obj, "name", None)
394 if name and isinstance(name, str) and name[0] != "<" and name[-1] != ">":
395 return Path(name).name
396 return default
399not_qtext_re = re.compile(r"[^\041\043-\133\135-\176]")
400QCONTENT = {chr(i) for i in range(0x20, 0x7F)} | {"\t"}
403def quoted_string(content: str) -> str:
404 """Return 7-bit content as quoted-string.
406 Format content into a quoted-string as defined in RFC5322 for
407 Internet Message Format. Notice that this is not the 8-bit HTTP
408 format, but the 7-bit email format. Content must be in usascii or
409 a ValueError is raised.
410 """
411 if not (QCONTENT > set(content)):
412 raise ValueError(f"bad content for quoted-string {content!r}")
413 return not_qtext_re.sub(lambda x: "\\" + x.group(0), content)
416def content_disposition_header(
417 disptype: str,
418 quote_fields: bool = True,
419 _charset: str = "utf-8",
420 params: dict[str, str] | None = None,
421) -> str:
422 """Sets ``Content-Disposition`` header for MIME.
424 This is the MIME payload Content-Disposition header from RFC 2183
425 and RFC 7579 section 4.2, not the HTTP Content-Disposition from
426 RFC 6266.
428 disptype is a disposition type: inline, attachment, form-data.
429 Should be valid extension token (see RFC 2183)
431 quote_fields performs value quoting to 7-bit MIME headers
432 according to RFC 7578. Set to quote_fields to False if recipient
433 can take 8-bit file names and field values.
435 _charset specifies the charset to use when quote_fields is True.
437 params is a dict with disposition params.
438 """
439 if not disptype or not (TOKEN > set(disptype)):
440 raise ValueError(f"bad content disposition type {disptype!r}")
442 value = disptype
443 if params:
444 lparams = []
445 for key, val in params.items():
446 if not key or not (TOKEN > set(key)):
447 raise ValueError(f"bad content disposition parameter {key!r}={val!r}")
448 if quote_fields:
449 if key.lower() == "filename":
450 qval = quote(val, "", encoding=_charset)
451 lparams.append((key, '"%s"' % qval))
452 else:
453 try:
454 qval = quoted_string(val)
455 except ValueError:
456 qval = "".join(
457 (_charset, "''", quote(val, "", encoding=_charset))
458 )
459 lparams.append((key + "*", qval))
460 else:
461 lparams.append((key, '"%s"' % qval))
462 else:
463 qval = val.replace("\\", "\\\\").replace('"', '\\"')
464 lparams.append((key, '"%s"' % qval))
465 sparams = "; ".join("=".join(pair) for pair in lparams)
466 value = "; ".join((value, sparams))
467 return value
470def is_expected_content_type(
471 response_content_type: str, expected_content_type: str
472) -> bool:
473 """Checks if received content type is processable as an expected one.
475 Both arguments should be given without parameters.
476 """
477 if expected_content_type == "application/json":
478 return json_re.match(response_content_type) is not None
479 return expected_content_type in response_content_type
482def is_ip_address(host: str | None) -> bool:
483 """Check if host looks like an IP Address.
485 This check is only meant as a heuristic to ensure that
486 a host is not a domain name.
487 """
488 if not host:
489 return False
490 # For a host to be an ipv4 address, it must be all numeric.
491 # The host must contain a colon to be an IPv6 address.
492 return ":" in host or host.replace(".", "").isdigit()
495def is_canonical_ipv4_address(host: str) -> bool:
496 """Check if host is a canonical dotted-quad IPv4 address.
498 Rejects the legacy numeric forms that ``socket`` still accepts and
499 maps onto an address, e.g. ``2130706433``, ``017700000001``, ``127.1``.
500 """
501 parts = host.split(".")
502 if len(parts) != 4:
503 return False
504 for part in parts:
505 # Each octet must be 1-3 ASCII digits; reject unicode digits
506 # (which ``str.isdigit`` accepts but ``int`` may not), octal
507 # leading zeros, and values above 255.
508 if not (1 <= len(part) <= 3) or not part.isascii() or not part.isdigit():
509 return False
510 if part[0] == "0" and len(part) != 1:
511 return False
512 if int(part) > 255:
513 return False
514 return True
517_cached_current_datetime: int | None = None
518_cached_formatted_datetime = ""
521def rfc822_formatted_time() -> str:
522 global _cached_current_datetime
523 global _cached_formatted_datetime
525 now = int(time.time())
526 if now != _cached_current_datetime:
527 # Weekday and month names for HTTP date/time formatting;
528 # always English!
529 # Tuples are constants stored in codeobject!
530 _weekdayname = ("Mon", "Tue", "Wed", "Thu", "Fri", "Sat", "Sun")
531 _monthname = (
532 "", # Dummy so we can use 1-based month numbers
533 "Jan",
534 "Feb",
535 "Mar",
536 "Apr",
537 "May",
538 "Jun",
539 "Jul",
540 "Aug",
541 "Sep",
542 "Oct",
543 "Nov",
544 "Dec",
545 )
547 year, month, day, hh, mm, ss, wd, *tail = time.gmtime(now)
548 _cached_formatted_datetime = "%s, %02d %3s %4d %02d:%02d:%02d GMT" % (
549 _weekdayname[wd],
550 day,
551 _monthname[month],
552 year,
553 hh,
554 mm,
555 ss,
556 )
557 _cached_current_datetime = now
558 return _cached_formatted_datetime
561def _weakref_handle(info: "tuple[weakref.ref[object], str]") -> None:
562 ref, name = info
563 ob = ref()
564 if ob is not None:
565 with suppress(Exception):
566 getattr(ob, name)()
569def weakref_handle(
570 ob: object,
571 name: str,
572 timeout: float | None,
573 loop: asyncio.AbstractEventLoop,
574 timeout_ceil_threshold: float = 5,
575) -> asyncio.TimerHandle | None:
576 if timeout is not None and timeout > 0:
577 when = loop.time() + timeout
578 if timeout >= timeout_ceil_threshold:
579 when = ceil(when)
581 return loop.call_at(when, _weakref_handle, (weakref.ref(ob), name))
582 return None
585def call_later(
586 cb: Callable[[], Any],
587 timeout: float | None,
588 loop: asyncio.AbstractEventLoop,
589 timeout_ceil_threshold: float = 5,
590) -> asyncio.TimerHandle | None:
591 if timeout is None or timeout <= 0:
592 return None
593 now = loop.time()
594 when = calculate_timeout_when(now, timeout, timeout_ceil_threshold)
595 return loop.call_at(when, cb)
598def calculate_timeout_when(
599 loop_time: float,
600 timeout: float,
601 timeout_ceiling_threshold: float,
602) -> float:
603 """Calculate when to execute a timeout."""
604 when = loop_time + timeout
605 if timeout > timeout_ceiling_threshold:
606 return ceil(when)
607 return when
610class TimeoutHandle:
611 """Timeout handle"""
613 __slots__ = ("_timeout", "_loop", "_ceil_threshold", "_callbacks")
615 def __init__(
616 self,
617 loop: asyncio.AbstractEventLoop,
618 timeout: float | None,
619 ceil_threshold: float = 5,
620 ) -> None:
621 self._timeout = timeout
622 self._loop = loop
623 self._ceil_threshold = ceil_threshold
624 self._callbacks: list[
625 tuple[Callable[..., None], tuple[Any, ...], dict[str, Any]]
626 ] = []
628 def register(
629 self, callback: Callable[..., None], *args: Any, **kwargs: Any
630 ) -> None:
631 self._callbacks.append((callback, args, kwargs))
633 def close(self) -> None:
634 self._callbacks.clear()
636 def start(self) -> asyncio.TimerHandle | None:
637 timeout = self._timeout
638 if timeout is not None and timeout > 0:
639 when = self._loop.time() + timeout
640 if timeout >= self._ceil_threshold:
641 when = ceil(when)
642 return self._loop.call_at(when, self.__call__)
643 else:
644 return None
646 def timer(self) -> "BaseTimerContext":
647 if self._timeout is not None and self._timeout > 0:
648 timer = TimerContext(self._loop)
649 self.register(timer.timeout)
650 return timer
651 else:
652 return TimerNoop()
654 def __call__(self) -> None:
655 for cb, args, kwargs in self._callbacks:
656 with suppress(Exception):
657 cb(*args, **kwargs)
659 self._callbacks.clear()
662class BaseTimerContext(ContextManager["BaseTimerContext"]):
664 __slots__ = ()
666 def assert_timeout(self) -> None:
667 """Raise TimeoutError if timeout has been exceeded."""
670class TimerNoop(BaseTimerContext):
672 __slots__ = ()
674 def __enter__(self) -> BaseTimerContext:
675 return self
677 def __exit__(
678 self,
679 exc_type: type[BaseException] | None,
680 exc_val: BaseException | None,
681 exc_tb: TracebackType | None,
682 ) -> None:
683 return
686class TimerContext(BaseTimerContext):
687 """Low resolution timeout context manager"""
689 __slots__ = ("_loop", "_tasks", "_cancelled", "_cancelling")
691 def __init__(self, loop: asyncio.AbstractEventLoop) -> None:
692 self._loop = loop
693 self._tasks: list[asyncio.Task[Any]] = []
694 self._cancelled = False
695 self._cancelling = 0
697 def assert_timeout(self) -> None:
698 """Raise TimeoutError if timer has already been cancelled."""
699 if self._cancelled:
700 raise asyncio.TimeoutError from None
702 def __enter__(self) -> BaseTimerContext:
703 task = asyncio.current_task(loop=self._loop)
704 if task is None:
705 raise RuntimeError("Timeout context manager should be used inside a task")
707 if sys.version_info >= (3, 11):
708 # Remember if the task was already cancelling
709 # so when we __exit__ we can decide if we should
710 # raise asyncio.TimeoutError or let the cancellation propagate
711 self._cancelling = task.cancelling()
713 if self._cancelled:
714 raise asyncio.TimeoutError from None
716 self._tasks.append(task)
717 return self
719 def __exit__(
720 self,
721 exc_type: type[BaseException] | None,
722 exc_val: BaseException | None,
723 exc_tb: TracebackType | None,
724 ) -> bool | None:
725 enter_task: asyncio.Task[Any] | None = None
726 if self._tasks:
727 enter_task = self._tasks.pop()
729 if exc_type is asyncio.CancelledError and self._cancelled:
730 assert enter_task is not None
731 # The timeout was hit, and the task was cancelled
732 # so we need to uncancel the last task that entered the context manager
733 # since the cancellation should not leak out of the context manager
734 if sys.version_info >= (3, 11):
735 # If the task was already cancelling don't raise
736 # asyncio.TimeoutError and instead return None
737 # to allow the cancellation to propagate
738 if enter_task.uncancel() > self._cancelling:
739 return None
740 raise asyncio.TimeoutError from exc_val
741 return None
743 def timeout(self) -> None:
744 if not self._cancelled:
745 for task in set(self._tasks):
746 task.cancel()
748 self._cancelled = True
751def ceil_timeout(
752 delay: float | None, ceil_threshold: float = 5
753) -> async_timeout.Timeout:
754 if delay is None or delay <= 0:
755 return async_timeout.timeout(None)
757 loop = asyncio.get_running_loop()
758 now = loop.time()
759 when = now + delay
760 if delay > ceil_threshold:
761 when = ceil(when)
762 return async_timeout.timeout_at(when)
765class HeadersDictProxy(Mapping[str, str]):
766 def __init__(self, md: CIMultiDict[str]):
767 self._md = md
769 def getall(self, key: str) -> tuple[str, ...]:
770 val = self.get(key, "")
771 unescape = _QUOTED_PAIR_SUB.sub
772 values = []
773 for m in _LIST_ELEMENT_RE.finditer(val):
774 qs = m.group(1)
775 if qs is not None:
776 values.append(unescape(r"\1", qs))
777 else:
778 raw = m.group(2).strip()
779 if raw:
780 values.append(
781 _PROTECTED_RE.sub(lambda p: unescape(r"\1", p.group()), raw)
782 )
783 return tuple(values)
785 def __eq__(self, other: object) -> bool:
786 return self._md.__eq__(other)
788 def __getitem__(self, key: str) -> str:
789 return ", ".join(self._md.getall(key))
791 def __iter__(self) -> Iterator[str]:
792 # We need to deduplicate keys from MultiDict
793 # But, we also need to retain ordering
794 seen = set()
795 for k in self._md.__iter__():
796 if k in seen:
797 continue
798 seen.add(k)
799 yield k
801 def __len__(self) -> int:
802 return len(set(self._md.keys()))
804 def __repr__(self) -> str:
805 body = ", ".join(f"'{k}': {v!r}" for k, v in self.items())
806 return f"<{self.__class__.__name__}({body})>"
809class HeadersMixin:
810 """Mixin for handling headers."""
812 _headers: Mapping[str, str]
813 _content_type: str | None = None
814 _content_dict: dict[str, str] | None = None
815 _stored_content_type: str | None | _SENTINEL = sentinel
817 def _parse_content_type(self, raw: str | None) -> None:
818 self._stored_content_type = raw
819 if raw is None:
820 # default value according to RFC 2616
821 self._content_type = "application/octet-stream"
822 self._content_dict = {}
823 else:
824 content_type, content_mapping_proxy = parse_content_type(raw)
825 self._content_type = content_type
826 # _content_dict needs to be mutable so we can update it
827 self._content_dict = content_mapping_proxy.copy()
829 @property
830 def content_type(self) -> str:
831 """The value of content part for Content-Type HTTP header."""
832 raw = self._headers.get(hdrs.CONTENT_TYPE)
833 if self._stored_content_type != raw:
834 self._parse_content_type(raw)
835 assert self._content_type is not None
836 return self._content_type
838 @property
839 def charset(self) -> str | None:
840 """The value of charset part for Content-Type HTTP header."""
841 raw = self._headers.get(hdrs.CONTENT_TYPE)
842 if self._stored_content_type != raw:
843 self._parse_content_type(raw)
844 assert self._content_dict is not None
845 return self._content_dict.get("charset")
847 @property
848 def content_length(self) -> int | None:
849 """The value of Content-Length HTTP header."""
850 content_length = self._headers.get(hdrs.CONTENT_LENGTH)
851 return None if content_length is None else int(content_length)
854def set_result(fut: "asyncio.Future[_T]", result: _T) -> None:
855 if not fut.done():
856 fut.set_result(result)
859_EXC_SENTINEL = BaseException()
862class ErrorableProtocol(Protocol):
863 def set_exception(
864 self,
865 exc: type[BaseException] | BaseException,
866 exc_cause: BaseException = ...,
867 ) -> None: ...
870def set_exception(
871 fut: Union["asyncio.Future[_T]", ErrorableProtocol],
872 exc: type[BaseException] | BaseException,
873 exc_cause: BaseException = _EXC_SENTINEL,
874) -> None:
875 """Set future exception.
877 If the future is marked as complete, this function is a no-op.
879 :param exc_cause: An exception that is a direct cause of ``exc``.
880 Only set if provided.
881 """
882 if asyncio.isfuture(fut) and fut.done():
883 return
885 exc_is_sentinel = exc_cause is _EXC_SENTINEL
886 exc_causes_itself = exc is exc_cause
887 if not exc_is_sentinel and not exc_causes_itself:
888 exc.__cause__ = exc_cause
890 fut.set_exception(exc)
893@functools.total_ordering
894class BaseKey(Generic[_T]):
895 """Base for concrete context storage key classes.
897 Each storage is provided with its own sub-class for the sake of some additional type safety.
898 """
900 __slots__ = ("_name", "_t", "__orig_class__")
902 # This may be set by Python when instantiating with a generic type. We need to
903 # support this, in order to support types that are not concrete classes,
904 # like Iterable, which can't be passed as the second parameter to __init__.
905 __orig_class__: type[object]
907 # TODO(PY314): Change Type to TypeForm (this should resolve unreachable below).
908 def __init__(self, name: str, t: type[_T] | None = None):
909 # Prefix with module name to help deduplicate key names.
910 frame = inspect.currentframe()
911 while frame:
912 if frame.f_code.co_name == "<module>":
913 module: str = frame.f_globals["__name__"]
914 break
915 frame = frame.f_back
916 else:
917 raise RuntimeError("Failed to get module name.")
919 # https://github.com/python/mypy/issues/14209
920 self._name = module + "." + name # type: ignore[possibly-undefined]
921 self._t = t
923 def __lt__(self, other: object) -> bool:
924 if isinstance(other, BaseKey):
925 return self._name < other._name
926 return True # Order BaseKey above other types.
928 def __repr__(self) -> str:
929 t = self._t
930 if t is None:
931 with suppress(AttributeError):
932 # Set to type arg.
933 t = get_args(self.__orig_class__)[0]
935 if t is None:
936 t_repr = "<<Unknown>>"
937 elif isinstance(t, type):
938 if t.__module__ == "builtins":
939 t_repr = t.__qualname__
940 else:
941 t_repr = f"{t.__module__}.{t.__qualname__}"
942 else:
943 t_repr = repr(t) # type: ignore[unreachable]
944 return f"<{self.__class__.__name__}({self._name}, type={t_repr})>"
947class AppKey(BaseKey[_T]):
948 """Keys for static typing support in Application."""
951class RequestKey(BaseKey[_T]):
952 """Keys for static typing support in Request."""
955class ResponseKey(BaseKey[_T]):
956 """Keys for static typing support in Response."""
959@final
960class ChainMapProxy(Mapping[str | AppKey[Any], Any]):
961 __slots__ = ("_maps",)
963 def __init__(self, maps: Iterable[Mapping[str | AppKey[Any], Any]]) -> None:
964 self._maps = tuple(maps)
966 def __init_subclass__(cls) -> None:
967 raise TypeError(
968 f"Inheritance class {cls.__name__} from ChainMapProxy is forbidden"
969 )
971 @overload # type: ignore[override]
972 def __getitem__(self, key: AppKey[_T]) -> _T: ...
974 @overload
975 def __getitem__(self, key: str) -> Any: ...
977 def __getitem__(self, key: str | AppKey[_T]) -> Any:
978 for mapping in self._maps:
979 try:
980 return mapping[key]
981 except KeyError:
982 pass
983 raise KeyError(key)
985 @overload # type: ignore[override]
986 def get(self, key: AppKey[_T], default: _S) -> _T | _S: ...
988 @overload
989 def get(self, key: AppKey[_T], default: None = ...) -> _T | None: ...
991 @overload
992 def get(self, key: str, default: Any = ...) -> Any: ...
994 def get(self, key: str | AppKey[_T], default: Any = None) -> Any:
995 try:
996 return self[key]
997 except KeyError:
998 return default
1000 def __len__(self) -> int:
1001 # reuses stored hash values if possible
1002 return len(set().union(*self._maps))
1004 def __iter__(self) -> Iterator[str | AppKey[Any]]:
1005 d: dict[str | AppKey[Any], Any] = {}
1006 for mapping in reversed(self._maps):
1007 # reuses stored hash values if possible
1008 d.update(mapping)
1009 return iter(d)
1011 def __contains__(self, key: object) -> bool:
1012 return any(key in m for m in self._maps)
1014 def __bool__(self) -> bool:
1015 return any(self._maps)
1017 def __repr__(self) -> str:
1018 content = ", ".join(map(repr, self._maps))
1019 return f"ChainMapProxy({content})"
1022class CookieMixin:
1023 """Mixin for handling cookies."""
1025 _cookies: SimpleCookie | None = None
1027 @property
1028 def cookies(self) -> SimpleCookie:
1029 if self._cookies is None:
1030 self._cookies = SimpleCookie()
1031 return self._cookies
1033 def set_cookie(
1034 self,
1035 name: str,
1036 value: str,
1037 *,
1038 expires: str | None = None,
1039 domain: str | None = None,
1040 max_age: int | str | None = None,
1041 path: str = "/",
1042 secure: bool | None = None,
1043 httponly: bool | None = None,
1044 samesite: str | None = None,
1045 partitioned: bool | None = None,
1046 ) -> None:
1047 """Set or update response cookie.
1049 Sets new cookie or updates existent with new value.
1050 Also updates only those params which are not None.
1051 """
1052 if self._cookies is None:
1053 self._cookies = SimpleCookie()
1055 self._cookies[name] = value
1056 c = self._cookies[name]
1058 if expires is not None:
1059 c["expires"] = expires
1060 elif c.get("expires") == "Thu, 01 Jan 1970 00:00:00 GMT":
1061 del c["expires"]
1063 if domain is not None:
1064 c["domain"] = domain
1066 if max_age is not None:
1067 c["max-age"] = str(max_age)
1068 elif "max-age" in c:
1069 del c["max-age"]
1071 c["path"] = path
1073 if secure is not None:
1074 c["secure"] = secure
1075 if httponly is not None:
1076 c["httponly"] = httponly
1077 if samesite is not None:
1078 c["samesite"] = samesite
1080 if partitioned is not None:
1081 c["partitioned"] = partitioned
1083 if DEBUG:
1084 cookie_length = len(c.output(header="")[1:])
1085 if cookie_length > COOKIE_MAX_LENGTH:
1086 warnings.warn(
1087 "The size of is too large, it might get ignored by the client.",
1088 UserWarning,
1089 stacklevel=2,
1090 )
1092 def del_cookie(
1093 self,
1094 name: str,
1095 *,
1096 domain: str | None = None,
1097 path: str = "/",
1098 secure: bool | None = None,
1099 httponly: bool | None = None,
1100 samesite: str | None = None,
1101 ) -> None:
1102 """Delete cookie.
1104 Creates new empty expired cookie.
1105 """
1106 # TODO: do we need domain/path here?
1107 if self._cookies is not None:
1108 self._cookies.pop(name, None)
1109 self.set_cookie(
1110 name,
1111 "",
1112 max_age=0,
1113 expires="Thu, 01 Jan 1970 00:00:00 GMT",
1114 domain=domain,
1115 path=path,
1116 secure=secure,
1117 httponly=httponly,
1118 samesite=samesite,
1119 )
1122def populate_with_cookies(headers: "CIMultiDict[str]", cookies: SimpleCookie) -> None:
1123 for cookie in cookies.values():
1124 value = cookie.output(header="")[1:]
1125 headers.add(hdrs.SET_COOKIE, value)
1128# https://tools.ietf.org/html/rfc7232#section-2.3
1129_ETAGC = r"[!\x23-\x7E\x80-\xff]+"
1130_ETAGC_RE = re.compile(_ETAGC)
1131_QUOTED_ETAG = rf'(W/)?"({_ETAGC})"'
1132QUOTED_ETAG_RE = re.compile(_QUOTED_ETAG)
1133LIST_QUOTED_ETAG_RE = re.compile(rf"({_QUOTED_ETAG})(?:\s*,\s*|$)|(.)")
1135ETAG_ANY = "*"
1138@frozen_dataclass_decorator
1139class ETag:
1140 value: str
1141 is_weak: bool = False
1144def validate_etag_value(value: str) -> None:
1145 if value != ETAG_ANY and not _ETAGC_RE.fullmatch(value):
1146 raise ValueError(
1147 f"Value {value!r} is not a valid etag. Maybe it contains '\"'?"
1148 )
1151def parse_http_date(date_str: str | None) -> datetime.datetime | None:
1152 """Process a date string, return a datetime object"""
1153 if date_str is not None:
1154 timetuple = parsedate(date_str)
1155 if timetuple is not None:
1156 with suppress(ValueError):
1157 return datetime.datetime(*timetuple[:6], tzinfo=datetime.timezone.utc)
1158 return None
1161@functools.lru_cache
1162def must_be_empty_body(method: str, code: int) -> bool:
1163 """Check if a request must return an empty body."""
1164 return (
1165 code in EMPTY_BODY_STATUS_CODES
1166 or method in EMPTY_BODY_METHODS
1167 or (200 <= code < 300 and method == hdrs.METH_CONNECT)
1168 )
1171def should_remove_content_length(method: str, code: int) -> bool:
1172 """Check if a Content-Length header should be removed.
1174 This should always be a subset of must_be_empty_body
1175 """
1176 # https://www.rfc-editor.org/rfc/rfc9110.html#section-8.6-8
1177 # https://www.rfc-editor.org/rfc/rfc9110.html#section-15.4.5-4
1178 return code in EMPTY_BODY_STATUS_CODES or (
1179 200 <= code < 300 and method == hdrs.METH_CONNECT
1180 )