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"\\(.)")
72if sys.version_info >= (3, 11):
73 _QUOTED_STRING_CONTENT = r'(?:[^"\\]++|\\.)*+'
74 _ESCAPED_COMMENT = r"(?:[^()\\]++|\\.)*+"
75 _LIST_ELEMENT = rf"""
76 [ \t]*
77 (?:
78 "( {_QUOTED_STRING_CONTENT} )" # group 1: top-level quoted-string
79 [ \t]* (?:,|\Z)
80 | ( # group 2: unquoted element
81 (?:
82 (?<=[^\s]=) "{_QUOTED_STRING_CONTENT}" # parameter quoted value
83 | (?<=\s) \( {_ESCAPED_COMMENT} \) # comment
84 | [^,"(\\]++ # run of ordinary characters
85 | [^,] # quote, paren or backslash the branches above rejected
86 )++
87 )
88 (?:,|\Z)
89 )
90 """
91else:
92 _QUOTED_STRING_CONTENT = r'(?:[^"\\]|\\.)*'
93 _ESCAPED_COMMENT = r"(?:[^()\\]|\\.)*"
94 _LIST_ELEMENT = rf"""
95 [ \t]*
96 (?:
97 "( {_QUOTED_STRING_CONTENT} )" # group 1: top-level quoted-string
98 [ \t]* (?:,|\Z)
99 | ( # group 2: unquoted element
100 (?:
101 (?<=[^\s]=) "{_QUOTED_STRING_CONTENT}" # parameter quoted value
102 | (?<=\s) \( {_ESCAPED_COMMENT} \) # comment
103 | [^,] # any non-comma character
104 )+?
105 )
106 (?:,|\Z)
107 )
108 """
109# Matches one element in a comma-separated header list.
110# Group 1: content of a top-level quoted-string (quotes stripped).
111# Group 2: an unquoted element (may contain parameter quoted-strings / comments).
112_LIST_ELEMENT_RE = re.compile(_LIST_ELEMENT, re.VERBOSE)
113# Finds parameter quoted-strings and comments inside an unquoted element for unescaping.
114_PROTECTED_RE = re.compile(
115 rf"""
116 (?<=[^\s]=) "{_QUOTED_STRING_CONTENT}" # parameter quoted-string
117 | (?<=\s) \( {_ESCAPED_COMMENT} \) # comment
118 """,
119 re.VERBOSE,
120)
122_T = TypeVar("_T")
123_S = TypeVar("_S")
125_SENTINEL = enum.Enum("_SENTINEL", "sentinel")
126sentinel = _SENTINEL.sentinel
128NO_EXTENSIONS = bool(os.environ.get("AIOHTTP_NO_EXTENSIONS"))
130# https://datatracker.ietf.org/doc/html/rfc9112#section-6.3-2.1
131EMPTY_BODY_STATUS_CODES = frozenset((204, 304, *range(100, 200)))
132# https://datatracker.ietf.org/doc/html/rfc9112#section-6.3-2.1
133# https://datatracker.ietf.org/doc/html/rfc9112#section-6.3-2.2
134EMPTY_BODY_METHODS = frozenset({hdrs.METH_HEAD})
136DEBUG = sys.flags.dev_mode or (
137 not sys.flags.ignore_environment and bool(os.environ.get("PYTHONASYNCIODEBUG"))
138)
141EMPTY_SCHEMA_SET = frozenset({""})
142HTTP_SCHEMA_SET = frozenset({"http", "https"})
143WS_SCHEMA_SET = frozenset({"ws", "wss"})
144HTTP_AND_EMPTY_SCHEMA_SET = HTTP_SCHEMA_SET | EMPTY_SCHEMA_SET
145HIGH_LEVEL_SCHEMA_SET = HTTP_AND_EMPTY_SCHEMA_SET | WS_SCHEMA_SET
148CHAR = {chr(i) for i in range(0, 128)}
149CTL = {chr(i) for i in range(0, 32)} | {
150 chr(127),
151}
152SEPARATORS = {
153 "(",
154 ")",
155 "<",
156 ">",
157 "@",
158 ",",
159 ";",
160 ":",
161 "\\",
162 '"',
163 "/",
164 "[",
165 "]",
166 "?",
167 "=",
168 "{",
169 "}",
170 " ",
171 chr(9),
172}
173TOKEN = CHAR ^ CTL ^ SEPARATORS
176json_re = re.compile(r"^(?:application/|[\w.-]+/[\w.+-]+?\+)json$", re.IGNORECASE)
179def encode_basic_auth(login: str, password: str = "", encoding: str = "utf-8") -> str:
180 """Encode HTTP Basic Authentication credentials as an Authorization header value.
182 Returns a string of the form ``"Basic <base64>"`` suitable for use as the
183 value of the ``Authorization`` (or ``Proxy-Authorization``) header.
184 """
185 if ":" in login:
186 raise ValueError('A ":" is not allowed in login (RFC 7617#section-2)')
187 creds = f"{login}:{password}".encode(encoding)
188 return "Basic " + base64.b64encode(creds).decode(encoding)
191def strip_auth_from_url(url: URL) -> tuple[URL, str | None]:
192 """Strip user/password from a URL and return the Authorization header value.
194 Returns a tuple of ``(url_without_credentials, authorization_header_value)``.
195 The header value is ``None`` if no credentials were present.
196 """
197 # Check raw_user and raw_password first as yarl is likely
198 # to already have these values parsed from the netloc in the cache.
199 if url.raw_user is None and url.raw_password is None:
200 return url, None
201 return url.with_user(None), encode_basic_auth(url.user or "", url.password or "")
204def netrc_from_env() -> netrc.netrc | None:
205 """Load netrc from file.
207 Attempt to load it from the path specified by the env-var
208 NETRC or in the default location in the user's home directory.
210 Returns None if it couldn't be found or fails to parse.
211 """
212 netrc_env = os.environ.get("NETRC")
214 if netrc_env is not None:
215 netrc_path = Path(netrc_env)
216 else:
217 try:
218 home_dir = Path.home()
219 except RuntimeError as e:
220 # if pathlib can't resolve home, it may raise a RuntimeError
221 client_logger.debug(
222 "Could not resolve home directory when "
223 "trying to look for .netrc file: %s",
224 e,
225 )
226 return None
228 netrc_path = home_dir / (
229 "_netrc" if platform.system() == "Windows" else ".netrc"
230 )
232 try:
233 return netrc.netrc(str(netrc_path))
234 except netrc.NetrcParseError as e:
235 client_logger.warning("Could not parse .netrc file: %s", e)
236 except OSError as e:
237 netrc_exists = False
238 with contextlib.suppress(OSError):
239 netrc_exists = netrc_path.is_file()
240 # we couldn't read the file (doesn't exist, permissions, etc.)
241 if netrc_env or netrc_exists:
242 # only warn if the environment wanted us to load it,
243 # or it appears like the default file does actually exist
244 client_logger.warning("Could not read .netrc file: %s", e)
246 return None
249@frozen_dataclass_decorator
250class ProxyInfo:
251 proxy: URL
252 proxy_auth: str | None
255def _auth_header_from_netrc(netrc_obj: netrc.netrc | None, host: str) -> str:
256 """Return a ``Proxy-Authorization`` header value for ``host`` from netrc.
258 :raises LookupError: if ``netrc_obj`` is :py:data:`None` or if no
259 entry is found for the ``host``.
260 """
261 if netrc_obj is None:
262 raise LookupError("No .netrc file found")
263 auth_from_netrc = netrc_obj.authenticators(host)
265 if auth_from_netrc is None:
266 raise LookupError(f"No entry for {host!s} found in the `.netrc` file.")
267 login, account, password = auth_from_netrc
269 # TODO(PY311): username = login or account
270 # Up to python 3.10, account could be None if not specified,
271 # and login will be empty string if not specified. From 3.11,
272 # login and account will be empty string if not specified.
273 username = login if (login or account is None) else account
275 # TODO(PY311): Remove this, as password will be empty string
276 # if not specified
277 if password is None:
278 password = "" # type: ignore[unreachable]
280 return encode_basic_auth(username, password)
283def proxies_from_env() -> dict[str, ProxyInfo]:
284 proxy_urls = {
285 k: URL(v)
286 for k, v in getproxies().items()
287 if k in ("http", "https", "ws", "wss")
288 }
289 netrc_obj = netrc_from_env()
290 stripped = {k: strip_auth_from_url(v) for k, v in proxy_urls.items()}
291 ret = {}
292 for proto, val in stripped.items():
293 proxy, auth = val
294 if proxy.scheme in ("https", "wss"):
295 client_logger.warning(
296 "%s proxies %s are not supported, ignoring", proxy.scheme.upper(), proxy
297 )
298 continue
299 if netrc_obj and auth is None:
300 if proxy.host is not None:
301 try:
302 auth = _auth_header_from_netrc(netrc_obj, proxy.host)
303 except LookupError:
304 auth = None
305 ret[proto] = ProxyInfo(proxy, auth)
306 return ret
309def get_env_proxy_for_url(url: URL) -> tuple[URL, str | None]:
310 """Get a permitted proxy for the given URL from the env."""
311 if url.host is not None and proxy_bypass(url.host):
312 raise LookupError(f"Proxying is disallowed for `{url.host!r}`")
314 proxies_in_env = proxies_from_env()
315 try:
316 proxy_info = proxies_in_env[url.scheme]
317 except KeyError:
318 raise LookupError(f"No proxies found for `{url!s}` in the env")
319 else:
320 return proxy_info.proxy, proxy_info.proxy_auth
323@frozen_dataclass_decorator
324class MimeType:
325 type: str
326 subtype: str
327 suffix: str
328 parameters: "MultiDictProxy[str]"
331@functools.lru_cache(maxsize=56)
332def parse_mimetype(mimetype: str) -> MimeType:
333 """Parses a MIME type into its components.
335 mimetype is a MIME type string.
337 Returns a MimeType object.
339 Example:
341 >>> parse_mimetype('text/html; charset=utf-8')
342 MimeType(type='text', subtype='html', suffix='',
343 parameters={'charset': 'utf-8'})
345 """
346 if not mimetype:
347 return MimeType(
348 type="", subtype="", suffix="", parameters=MultiDictProxy(MultiDict())
349 )
351 parts = mimetype.split(";")
352 params: MultiDict[str] = MultiDict()
353 for item in parts[1:]:
354 if not item.strip():
355 continue
356 key, _, value = item.partition("=")
357 params.add(key.lower().strip(), value.strip(' "'))
359 fulltype = parts[0].strip().lower()
360 if fulltype == "*":
361 fulltype = "*/*"
363 mtype, _, stype = fulltype.partition("/")
364 stype, _, suffix = stype.partition("+")
366 return MimeType(
367 type=mtype, subtype=stype, suffix=suffix, parameters=MultiDictProxy(params)
368 )
371class EnsureOctetStream(EmailMessage):
372 def __init__(self) -> None:
373 super().__init__()
374 # https://www.rfc-editor.org/rfc/rfc9110#section-8.3-5
375 self.set_default_type("application/octet-stream")
377 def get_content_type(self) -> str:
378 """Re-implementation from Message
380 Returns application/octet-stream in place of plain/text when
381 value is wrong.
383 The way this class is used guarantees that content-type will
384 be present so simplify the checks wrt to the base implementation.
385 """
386 value = self.get("content-type", "").lower()
388 # Based on the implementation of _splitparam in the standard library
389 ctype, _, _ = value.partition(";")
390 ctype = ctype.strip()
391 if ctype.count("/") != 1:
392 return self.get_default_type()
393 return ctype
396@functools.lru_cache(maxsize=56)
397def parse_content_type(raw: str) -> tuple[str, MappingProxyType[str, str]]:
398 """Parse Content-Type header.
400 Returns a tuple of the parsed content type and a
401 MappingProxyType of parameters. The default returned value
402 is `application/octet-stream`
403 """
404 msg = HeaderParser(EnsureOctetStream, policy=HTTP).parsestr(f"Content-Type: {raw}")
405 content_type = msg.get_content_type()
406 params = msg.get_params(())
407 content_dict = dict(params[1:]) # First element is content type again
408 return content_type, MappingProxyType(content_dict)
411def guess_filename(obj: Any, default: str | None = None) -> str | None:
412 name = getattr(obj, "name", None)
413 if name and isinstance(name, str) and name[0] != "<" and name[-1] != ">":
414 return Path(name).name
415 return default
418not_qtext_re = re.compile(r"[^\041\043-\133\135-\176]")
419QCONTENT = {chr(i) for i in range(0x20, 0x7F)} | {"\t"}
422def quoted_string(content: str) -> str:
423 """Return 7-bit content as quoted-string.
425 Format content into a quoted-string as defined in RFC5322 for
426 Internet Message Format. Notice that this is not the 8-bit HTTP
427 format, but the 7-bit email format. Content must be in usascii or
428 a ValueError is raised.
429 """
430 if not (QCONTENT > set(content)):
431 raise ValueError(f"bad content for quoted-string {content!r}")
432 return not_qtext_re.sub(lambda x: "\\" + x.group(0), content)
435def content_disposition_header(
436 disptype: str,
437 quote_fields: bool = True,
438 _charset: str = "utf-8",
439 params: dict[str, str] | None = None,
440) -> str:
441 """Sets ``Content-Disposition`` header for MIME.
443 This is the MIME payload Content-Disposition header from RFC 2183
444 and RFC 7579 section 4.2, not the HTTP Content-Disposition from
445 RFC 6266.
447 disptype is a disposition type: inline, attachment, form-data.
448 Should be valid extension token (see RFC 2183)
450 quote_fields performs value quoting to 7-bit MIME headers
451 according to RFC 7578. Set to quote_fields to False if recipient
452 can take 8-bit file names and field values.
454 _charset specifies the charset to use when quote_fields is True.
456 params is a dict with disposition params.
457 """
458 if not disptype or not (TOKEN > set(disptype)):
459 raise ValueError(f"bad content disposition type {disptype!r}")
461 value = disptype
462 if params:
463 lparams = []
464 for key, val in params.items():
465 if not key or not (TOKEN > set(key)):
466 raise ValueError(f"bad content disposition parameter {key!r}={val!r}")
467 if quote_fields:
468 if key.lower() == "filename":
469 qval = quote(val, "", encoding=_charset)
470 lparams.append((key, '"%s"' % qval))
471 else:
472 try:
473 qval = quoted_string(val)
474 except ValueError:
475 qval = "".join(
476 (_charset, "''", quote(val, "", encoding=_charset))
477 )
478 lparams.append((key + "*", qval))
479 else:
480 lparams.append((key, '"%s"' % qval))
481 else:
482 qval = val.replace("\\", "\\\\").replace('"', '\\"')
483 lparams.append((key, '"%s"' % qval))
484 sparams = "; ".join("=".join(pair) for pair in lparams)
485 value = "; ".join((value, sparams))
486 return value
489def is_expected_content_type(
490 response_content_type: str, expected_content_type: str
491) -> bool:
492 """Checks if received content type is processable as an expected one.
494 Both arguments should be given without parameters.
495 """
496 if expected_content_type == "application/json":
497 return json_re.match(response_content_type) is not None
498 return expected_content_type in response_content_type
501def is_ip_address(host: str | None) -> bool:
502 """Check if host looks like an IP Address.
504 This check is only meant as a heuristic to ensure that
505 a host is not a domain name.
506 """
507 if not host:
508 return False
509 # For a host to be an ipv4 address, it must be all numeric.
510 # The host must contain a colon to be an IPv6 address.
511 return ":" in host or host.replace(".", "").isdigit()
514def is_canonical_ipv4_address(host: str) -> bool:
515 """Check if host is a canonical dotted-quad IPv4 address.
517 Rejects the legacy numeric forms that ``socket`` still accepts and
518 maps onto an address, e.g. ``2130706433``, ``017700000001``, ``127.1``.
519 """
520 parts = host.split(".")
521 if len(parts) != 4:
522 return False
523 for part in parts:
524 # Each octet must be 1-3 ASCII digits; reject unicode digits
525 # (which ``str.isdigit`` accepts but ``int`` may not), octal
526 # leading zeros, and values above 255.
527 if not (1 <= len(part) <= 3) or not part.isascii() or not part.isdigit():
528 return False
529 if part[0] == "0" and len(part) != 1:
530 return False
531 if int(part) > 255:
532 return False
533 return True
536_cached_current_datetime: int | None = None
537_cached_formatted_datetime = ""
540def rfc822_formatted_time() -> str:
541 global _cached_current_datetime
542 global _cached_formatted_datetime
544 now = int(time.time())
545 if now != _cached_current_datetime:
546 # Weekday and month names for HTTP date/time formatting;
547 # always English!
548 # Tuples are constants stored in codeobject!
549 _weekdayname = ("Mon", "Tue", "Wed", "Thu", "Fri", "Sat", "Sun")
550 _monthname = (
551 "", # Dummy so we can use 1-based month numbers
552 "Jan",
553 "Feb",
554 "Mar",
555 "Apr",
556 "May",
557 "Jun",
558 "Jul",
559 "Aug",
560 "Sep",
561 "Oct",
562 "Nov",
563 "Dec",
564 )
566 year, month, day, hh, mm, ss, wd, *tail = time.gmtime(now)
567 _cached_formatted_datetime = "%s, %02d %3s %4d %02d:%02d:%02d GMT" % (
568 _weekdayname[wd],
569 day,
570 _monthname[month],
571 year,
572 hh,
573 mm,
574 ss,
575 )
576 _cached_current_datetime = now
577 return _cached_formatted_datetime
580def _weakref_handle(info: "tuple[weakref.ref[object], str]") -> None:
581 ref, name = info
582 ob = ref()
583 if ob is not None:
584 with suppress(Exception):
585 getattr(ob, name)()
588def weakref_handle(
589 ob: object,
590 name: str,
591 timeout: float | None,
592 loop: asyncio.AbstractEventLoop,
593 timeout_ceil_threshold: float = 5,
594) -> asyncio.TimerHandle | None:
595 if timeout is not None and timeout > 0:
596 when = loop.time() + timeout
597 if timeout >= timeout_ceil_threshold:
598 when = ceil(when)
600 return loop.call_at(when, _weakref_handle, (weakref.ref(ob), name))
601 return None
604def call_later(
605 cb: Callable[[], Any],
606 timeout: float | None,
607 loop: asyncio.AbstractEventLoop,
608 timeout_ceil_threshold: float = 5,
609) -> asyncio.TimerHandle | None:
610 if timeout is None or timeout <= 0:
611 return None
612 now = loop.time()
613 when = calculate_timeout_when(now, timeout, timeout_ceil_threshold)
614 return loop.call_at(when, cb)
617def calculate_timeout_when(
618 loop_time: float,
619 timeout: float,
620 timeout_ceiling_threshold: float,
621) -> float:
622 """Calculate when to execute a timeout."""
623 when = loop_time + timeout
624 if timeout > timeout_ceiling_threshold:
625 return ceil(when)
626 return when
629class TimeoutHandle:
630 """Timeout handle"""
632 __slots__ = ("_timeout", "_loop", "_ceil_threshold", "_callbacks")
634 def __init__(
635 self,
636 loop: asyncio.AbstractEventLoop,
637 timeout: float | None,
638 ceil_threshold: float = 5,
639 ) -> None:
640 self._timeout = timeout
641 self._loop = loop
642 self._ceil_threshold = ceil_threshold
643 self._callbacks: list[
644 tuple[Callable[..., None], tuple[Any, ...], dict[str, Any]]
645 ] = []
647 def register(
648 self, callback: Callable[..., None], *args: Any, **kwargs: Any
649 ) -> None:
650 self._callbacks.append((callback, args, kwargs))
652 def close(self) -> None:
653 self._callbacks.clear()
655 def start(self) -> asyncio.TimerHandle | None:
656 timeout = self._timeout
657 if timeout is not None and timeout > 0:
658 when = self._loop.time() + timeout
659 if timeout >= self._ceil_threshold:
660 when = ceil(when)
661 return self._loop.call_at(when, self.__call__)
662 else:
663 return None
665 def timer(self) -> "BaseTimerContext":
666 if self._timeout is not None and self._timeout > 0:
667 timer = TimerContext(self._loop)
668 self.register(timer.timeout)
669 return timer
670 else:
671 return TimerNoop()
673 def __call__(self) -> None:
674 for cb, args, kwargs in self._callbacks:
675 with suppress(Exception):
676 cb(*args, **kwargs)
678 self._callbacks.clear()
681class BaseTimerContext(ContextManager["BaseTimerContext"]):
683 __slots__ = ()
685 def assert_timeout(self) -> None:
686 """Raise TimeoutError if timeout has been exceeded."""
689class TimerNoop(BaseTimerContext):
691 __slots__ = ()
693 def __enter__(self) -> BaseTimerContext:
694 return self
696 def __exit__(
697 self,
698 exc_type: type[BaseException] | None,
699 exc_val: BaseException | None,
700 exc_tb: TracebackType | None,
701 ) -> None:
702 return
705class TimerContext(BaseTimerContext):
706 """Low resolution timeout context manager"""
708 __slots__ = ("_loop", "_tasks", "_cancelled", "_cancelling")
710 def __init__(self, loop: asyncio.AbstractEventLoop) -> None:
711 self._loop = loop
712 self._tasks: list[asyncio.Task[Any]] = []
713 self._cancelled = False
714 self._cancelling = 0
716 def assert_timeout(self) -> None:
717 """Raise TimeoutError if timer has already been cancelled."""
718 if self._cancelled:
719 raise asyncio.TimeoutError from None
721 def __enter__(self) -> BaseTimerContext:
722 task = asyncio.current_task(loop=self._loop)
723 if task is None:
724 raise RuntimeError("Timeout context manager should be used inside a task")
726 if sys.version_info >= (3, 11):
727 # Remember if the task was already cancelling
728 # so when we __exit__ we can decide if we should
729 # raise asyncio.TimeoutError or let the cancellation propagate
730 self._cancelling = task.cancelling()
732 if self._cancelled:
733 raise asyncio.TimeoutError from None
735 self._tasks.append(task)
736 return self
738 def __exit__(
739 self,
740 exc_type: type[BaseException] | None,
741 exc_val: BaseException | None,
742 exc_tb: TracebackType | None,
743 ) -> bool | None:
744 enter_task: asyncio.Task[Any] | None = None
745 if self._tasks:
746 enter_task = self._tasks.pop()
748 if exc_type is asyncio.CancelledError and self._cancelled:
749 assert enter_task is not None
750 # The timeout was hit, and the task was cancelled
751 # so we need to uncancel the last task that entered the context manager
752 # since the cancellation should not leak out of the context manager
753 if sys.version_info >= (3, 11):
754 # If the task was already cancelling don't raise
755 # asyncio.TimeoutError and instead return None
756 # to allow the cancellation to propagate
757 if enter_task.uncancel() > self._cancelling:
758 return None
759 raise asyncio.TimeoutError from exc_val
760 return None
762 def timeout(self) -> None:
763 if not self._cancelled:
764 for task in set(self._tasks):
765 task.cancel()
767 self._cancelled = True
770def ceil_timeout(
771 delay: float | None, ceil_threshold: float = 5
772) -> async_timeout.Timeout:
773 if delay is None or delay <= 0:
774 return async_timeout.timeout(None)
776 loop = asyncio.get_running_loop()
777 now = loop.time()
778 when = now + delay
779 if delay > ceil_threshold:
780 when = ceil(when)
781 return async_timeout.timeout_at(when)
784class HeadersDictProxy(Mapping[str, str]):
785 def __init__(self, md: CIMultiDict[str]):
786 self._md = md
788 def getall(self, key: str) -> tuple[str, ...]:
789 val = self.get(key, "")
790 unescape = _QUOTED_PAIR_SUB.sub
791 values = []
792 for m in _LIST_ELEMENT_RE.finditer(val):
793 qs = m.group(1)
794 if qs is not None:
795 values.append(unescape(r"\1", qs))
796 else:
797 raw = m.group(2).strip()
798 if raw:
799 values.append(
800 _PROTECTED_RE.sub(lambda p: unescape(r"\1", p.group()), raw)
801 )
802 return tuple(values)
804 def __eq__(self, other: object) -> bool:
805 return self._md.__eq__(other)
807 def __getitem__(self, key: str) -> str:
808 return ", ".join(self._md.getall(key))
810 def __iter__(self) -> Iterator[str]:
811 # We need to deduplicate keys from MultiDict
812 # But, we also need to retain ordering
813 seen = set()
814 for k in self._md.__iter__():
815 if k in seen:
816 continue
817 seen.add(k)
818 yield k
820 def __len__(self) -> int:
821 return len(set(self._md.keys()))
823 def __repr__(self) -> str:
824 body = ", ".join(f"'{k}': {v!r}" for k, v in self.items())
825 return f"<{self.__class__.__name__}({body})>"
828class HeadersMixin:
829 """Mixin for handling headers."""
831 _headers: Mapping[str, str]
832 _content_type: str | None = None
833 _content_dict: dict[str, str] | None = None
834 _stored_content_type: str | None | _SENTINEL = sentinel
836 def _parse_content_type(self, raw: str | None) -> None:
837 self._stored_content_type = raw
838 if raw is None:
839 # default value according to RFC 2616
840 self._content_type = "application/octet-stream"
841 self._content_dict = {}
842 else:
843 content_type, content_mapping_proxy = parse_content_type(raw)
844 self._content_type = content_type
845 # _content_dict needs to be mutable so we can update it
846 self._content_dict = content_mapping_proxy.copy()
848 @property
849 def content_type(self) -> str:
850 """The value of content part for Content-Type HTTP header."""
851 raw = self._headers.get(hdrs.CONTENT_TYPE)
852 if self._stored_content_type != raw:
853 self._parse_content_type(raw)
854 assert self._content_type is not None
855 return self._content_type
857 @property
858 def charset(self) -> str | None:
859 """The value of charset part for Content-Type HTTP header."""
860 raw = self._headers.get(hdrs.CONTENT_TYPE)
861 if self._stored_content_type != raw:
862 self._parse_content_type(raw)
863 assert self._content_dict is not None
864 return self._content_dict.get("charset")
866 @property
867 def content_length(self) -> int | None:
868 """The value of Content-Length HTTP header."""
869 content_length = self._headers.get(hdrs.CONTENT_LENGTH)
870 return None if content_length is None else int(content_length)
873def set_result(fut: "asyncio.Future[_T]", result: _T) -> None:
874 if not fut.done():
875 fut.set_result(result)
878_EXC_SENTINEL = BaseException()
881class ErrorableProtocol(Protocol):
882 def set_exception(
883 self,
884 exc: type[BaseException] | BaseException,
885 exc_cause: BaseException = ...,
886 ) -> None: ...
889def set_exception(
890 fut: Union["asyncio.Future[_T]", ErrorableProtocol],
891 exc: type[BaseException] | BaseException,
892 exc_cause: BaseException = _EXC_SENTINEL,
893) -> None:
894 """Set future exception.
896 If the future is marked as complete, this function is a no-op.
898 :param exc_cause: An exception that is a direct cause of ``exc``.
899 Only set if provided.
900 """
901 if asyncio.isfuture(fut) and fut.done():
902 return
904 exc_is_sentinel = exc_cause is _EXC_SENTINEL
905 exc_causes_itself = exc is exc_cause
906 if not exc_is_sentinel and not exc_causes_itself:
907 exc.__cause__ = exc_cause
909 fut.set_exception(exc)
912@functools.total_ordering
913class BaseKey(Generic[_T]):
914 """Base for concrete context storage key classes.
916 Each storage is provided with its own sub-class for the sake of some additional type safety.
917 """
919 __slots__ = ("_name", "_t", "__orig_class__")
921 # This may be set by Python when instantiating with a generic type. We need to
922 # support this, in order to support types that are not concrete classes,
923 # like Iterable, which can't be passed as the second parameter to __init__.
924 __orig_class__: type[object]
926 # TODO(PY314): Change Type to TypeForm (this should resolve unreachable below).
927 def __init__(self, name: str, t: type[_T] | None = None):
928 # Prefix with module name to help deduplicate key names.
929 frame = inspect.currentframe()
930 while frame:
931 if frame.f_code.co_name == "<module>":
932 module: str = frame.f_globals["__name__"]
933 break
934 frame = frame.f_back
935 else:
936 raise RuntimeError("Failed to get module name.")
938 # https://github.com/python/mypy/issues/14209
939 self._name = module + "." + name # type: ignore[possibly-undefined]
940 self._t = t
942 def __lt__(self, other: object) -> bool:
943 if isinstance(other, BaseKey):
944 return self._name < other._name
945 return True # Order BaseKey above other types.
947 def __repr__(self) -> str:
948 t = self._t
949 if t is None:
950 with suppress(AttributeError):
951 # Set to type arg.
952 t = get_args(self.__orig_class__)[0]
954 if t is None:
955 t_repr = "<<Unknown>>"
956 elif isinstance(t, type):
957 if t.__module__ == "builtins":
958 t_repr = t.__qualname__
959 else:
960 t_repr = f"{t.__module__}.{t.__qualname__}"
961 else:
962 t_repr = repr(t) # type: ignore[unreachable]
963 return f"<{self.__class__.__name__}({self._name}, type={t_repr})>"
966class AppKey(BaseKey[_T]):
967 """Keys for static typing support in Application."""
970class RequestKey(BaseKey[_T]):
971 """Keys for static typing support in Request."""
974class ResponseKey(BaseKey[_T]):
975 """Keys for static typing support in Response."""
978@final
979class ChainMapProxy(Mapping[str | AppKey[Any], Any]):
980 __slots__ = ("_maps",)
982 def __init__(self, maps: Iterable[Mapping[str | AppKey[Any], Any]]) -> None:
983 self._maps = tuple(maps)
985 def __init_subclass__(cls) -> None:
986 raise TypeError(
987 f"Inheritance class {cls.__name__} from ChainMapProxy is forbidden"
988 )
990 @overload # type: ignore[override]
991 def __getitem__(self, key: AppKey[_T]) -> _T: ...
993 @overload
994 def __getitem__(self, key: str) -> Any: ...
996 def __getitem__(self, key: str | AppKey[_T]) -> Any:
997 for mapping in self._maps:
998 try:
999 return mapping[key]
1000 except KeyError:
1001 pass
1002 raise KeyError(key)
1004 @overload # type: ignore[override]
1005 def get(self, key: AppKey[_T], default: _S) -> _T | _S: ...
1007 @overload
1008 def get(self, key: AppKey[_T], default: None = ...) -> _T | None: ...
1010 @overload
1011 def get(self, key: str, default: Any = ...) -> Any: ...
1013 def get(self, key: str | AppKey[_T], default: Any = None) -> Any:
1014 try:
1015 return self[key]
1016 except KeyError:
1017 return default
1019 def __len__(self) -> int:
1020 # reuses stored hash values if possible
1021 return len(set().union(*self._maps))
1023 def __iter__(self) -> Iterator[str | AppKey[Any]]:
1024 d: dict[str | AppKey[Any], Any] = {}
1025 for mapping in reversed(self._maps):
1026 # reuses stored hash values if possible
1027 d.update(mapping)
1028 return iter(d)
1030 def __contains__(self, key: object) -> bool:
1031 return any(key in m for m in self._maps)
1033 def __bool__(self) -> bool:
1034 return any(self._maps)
1036 def __repr__(self) -> str:
1037 content = ", ".join(map(repr, self._maps))
1038 return f"ChainMapProxy({content})"
1041class CookieMixin:
1042 """Mixin for handling cookies."""
1044 _cookies: SimpleCookie | None = None
1046 @property
1047 def cookies(self) -> SimpleCookie:
1048 if self._cookies is None:
1049 self._cookies = SimpleCookie()
1050 return self._cookies
1052 def set_cookie(
1053 self,
1054 name: str,
1055 value: str,
1056 *,
1057 expires: str | None = None,
1058 domain: str | None = None,
1059 max_age: int | str | None = None,
1060 path: str = "/",
1061 secure: bool | None = None,
1062 httponly: bool | None = None,
1063 samesite: str | None = None,
1064 partitioned: bool | None = None,
1065 ) -> None:
1066 """Set or update response cookie.
1068 Sets new cookie or updates existent with new value.
1069 Also updates only those params which are not None.
1070 """
1071 if self._cookies is None:
1072 self._cookies = SimpleCookie()
1074 self._cookies[name] = value
1075 c = self._cookies[name]
1077 if expires is not None:
1078 c["expires"] = expires
1079 elif c.get("expires") == "Thu, 01 Jan 1970 00:00:00 GMT":
1080 del c["expires"]
1082 if domain is not None:
1083 c["domain"] = domain
1085 if max_age is not None:
1086 c["max-age"] = str(max_age)
1087 elif "max-age" in c:
1088 del c["max-age"]
1090 c["path"] = path
1092 if secure is not None:
1093 c["secure"] = secure
1094 if httponly is not None:
1095 c["httponly"] = httponly
1096 if samesite is not None:
1097 c["samesite"] = samesite
1099 if partitioned is not None:
1100 c["partitioned"] = partitioned
1102 if DEBUG:
1103 cookie_length = len(c.output(header="")[1:])
1104 if cookie_length > COOKIE_MAX_LENGTH:
1105 warnings.warn(
1106 "The size of is too large, it might get ignored by the client.",
1107 UserWarning,
1108 stacklevel=2,
1109 )
1111 def del_cookie(
1112 self,
1113 name: str,
1114 *,
1115 domain: str | None = None,
1116 path: str = "/",
1117 secure: bool | None = None,
1118 httponly: bool | None = None,
1119 samesite: str | None = None,
1120 ) -> None:
1121 """Delete cookie.
1123 Creates new empty expired cookie.
1124 """
1125 # TODO: do we need domain/path here?
1126 if self._cookies is not None:
1127 self._cookies.pop(name, None)
1128 self.set_cookie(
1129 name,
1130 "",
1131 max_age=0,
1132 expires="Thu, 01 Jan 1970 00:00:00 GMT",
1133 domain=domain,
1134 path=path,
1135 secure=secure,
1136 httponly=httponly,
1137 samesite=samesite,
1138 )
1141def populate_with_cookies(headers: "CIMultiDict[str]", cookies: SimpleCookie) -> None:
1142 for cookie in cookies.values():
1143 value = cookie.output(header="")[1:]
1144 headers.add(hdrs.SET_COOKIE, value)
1147# https://tools.ietf.org/html/rfc7232#section-2.3
1148_ETAGC = r"[!\x23-\x7E\x80-\xff]+"
1149_ETAGC_RE = re.compile(_ETAGC)
1150_QUOTED_ETAG = rf'(W/)?"({_ETAGC})"'
1151QUOTED_ETAG_RE = re.compile(_QUOTED_ETAG)
1152LIST_QUOTED_ETAG_RE = re.compile(rf"({_QUOTED_ETAG})(?:\s*,\s*|$)|(.)")
1154ETAG_ANY = "*"
1157@frozen_dataclass_decorator
1158class ETag:
1159 value: str
1160 is_weak: bool = False
1163def validate_etag_value(value: str) -> None:
1164 if value != ETAG_ANY and not _ETAGC_RE.fullmatch(value):
1165 raise ValueError(
1166 f"Value {value!r} is not a valid etag. Maybe it contains '\"'?"
1167 )
1170def parse_http_date(date_str: str | None) -> datetime.datetime | None:
1171 """Process a date string, return a datetime object"""
1172 if date_str is not None:
1173 timetuple = parsedate(date_str)
1174 if timetuple is not None:
1175 with suppress(ValueError):
1176 return datetime.datetime(*timetuple[:6], tzinfo=datetime.timezone.utc)
1177 return None
1180@functools.lru_cache
1181def must_be_empty_body(method: str, code: int) -> bool:
1182 """Check if a request must return an empty body."""
1183 return (
1184 code in EMPTY_BODY_STATUS_CODES
1185 or method in EMPTY_BODY_METHODS
1186 or (200 <= code < 300 and method == hdrs.METH_CONNECT)
1187 )
1190def should_remove_content_length(method: str, code: int) -> bool:
1191 """Check if a Content-Length header should be removed.
1193 This should always be a subset of must_be_empty_body
1194 """
1195 # https://www.rfc-editor.org/rfc/rfc9110.html#section-8.6-8
1196 # https://www.rfc-editor.org/rfc/rfc9110.html#section-15.4.5-4
1197 return code in EMPTY_BODY_STATUS_CODES or (
1198 200 <= code < 300 and method == hdrs.METH_CONNECT
1199 )