Coverage for /pythoncovmergedfiles/medio/medio/src/aiohttp/aiohttp/helpers.py: 38%

1"""Various helper functions""" 

2 

3import asyncio 

4import base64 

5import binascii 

6import contextlib 

7import dataclasses 

8import datetime 

9import enum 

10import functools 

11import inspect 

12import netrc 

13import os 

14import platform 

15import re 

16import sys 

17import time 

18import warnings 

19import weakref 

20from collections import namedtuple 

21from contextlib import suppress 

22from email.parser import HeaderParser 

23from email.utils import parsedate 

24from http.cookies import SimpleCookie 

25from math import ceil 

26from pathlib import Path 

27from types import TracebackType 

28from typing import ( 

29 Any, 

30 Callable, 

31 ContextManager, 

32 Dict, 

33 Generator, 

34 Generic, 

35 Iterable, 

36 Iterator, 

37 List, 

38 Mapping, 

39 Optional, 

40 Pattern, 

41 Protocol, 

42 Tuple, 

43 Type, 

44 TypeVar, 

45 Union, 

46 final, 

47 get_args, 

48 overload, 

49) 

50from urllib.parse import quote 

51from urllib.request import getproxies, proxy_bypass 

52 

53from multidict import CIMultiDict, MultiDict, MultiDictProxy 

54from yarl import URL 

55 

56from . import hdrs 

57from .log import client_logger 

58from .typedefs import PathLike # noqa 

59 

60if sys.version_info >= (3, 11): 

61 import asyncio as async_timeout 

62else: 

63 import async_timeout 

64 

65__all__ = ("BasicAuth", "ChainMapProxy", "ETag") 

66 

67PY_310 = sys.version_info >= (3, 10) 

68 

69COOKIE_MAX_LENGTH = 4096 

70 

71_T = TypeVar("_T") 

72_S = TypeVar("_S") 

73 

74_SENTINEL = enum.Enum("_SENTINEL", "sentinel") 

75sentinel = _SENTINEL.sentinel 

76 

77NO_EXTENSIONS = bool(os.environ.get("AIOHTTP_NO_EXTENSIONS")) 

78 

79DEBUG = sys.flags.dev_mode or ( 

80 not sys.flags.ignore_environment and bool(os.environ.get("PYTHONASYNCIODEBUG")) 

81) 

82 

83 

84CHAR = {chr(i) for i in range(0, 128)} 

85CTL = {chr(i) for i in range(0, 32)} | { 

86 chr(127), 

87} 

88SEPARATORS = { 

89 "(", 

90 ")", 

91 "<", 

92 ">", 

93 "@", 

94 ",", 

95 ";", 

96 ":", 

97 "\\", 

98 '"', 

99 "/", 

100 "[", 

101 "]", 

102 "?", 

103 "=", 

104 "{", 

105 "}", 

106 " ", 

107 chr(9), 

108} 

109TOKEN = CHAR ^ CTL ^ SEPARATORS 

110 

111 

112class noop: 

113 def __await__(self) -> Generator[None, None, None]: 

114 yield 

115 

116 

117json_re = re.compile(r"(?:application/|[\w.-]+/[\w.+-]+?\+)json$", re.IGNORECASE) 

118 

119 

120class BasicAuth(namedtuple("BasicAuth", ["login", "password", "encoding"])): 

121 """Http basic authentication helper.""" 

122 

123 def __new__( 

124 cls, login: str, password: str = "", encoding: str = "latin1" 

125 ) -> "BasicAuth": 

126 if login is None: 

127 raise ValueError("None is not allowed as login value") 

128 

129 if password is None: 

130 raise ValueError("None is not allowed as password value") 

131 

132 if ":" in login: 

133 raise ValueError('A ":" is not allowed in login (RFC 1945#section-11.1)') 

134 

135 return super().__new__(cls, login, password, encoding) 

136 

137 @classmethod 

138 def decode(cls, auth_header: str, encoding: str = "latin1") -> "BasicAuth": 

139 """Create a BasicAuth object from an Authorization HTTP header.""" 

140 try: 

141 auth_type, encoded_credentials = auth_header.split(" ", 1) 

142 except ValueError: 

143 raise ValueError("Could not parse authorization header.") 

144 

145 if auth_type.lower() != "basic": 

146 raise ValueError("Unknown authorization method %s" % auth_type) 

147 

148 try: 

149 decoded = base64.b64decode( 

150 encoded_credentials.encode("ascii"), validate=True 

151 ).decode(encoding) 

152 except binascii.Error: 

153 raise ValueError("Invalid base64 encoding.") 

154 

155 try: 

156 # RFC 2617 HTTP Authentication 

157 # https://www.ietf.org/rfc/rfc2617.txt 

158 # the colon must be present, but the username and password may be 

159 # otherwise blank. 

160 username, password = decoded.split(":", 1) 

161 except ValueError: 

162 raise ValueError("Invalid credentials.") 

163 

164 return cls(username, password, encoding=encoding) 

165 

166 @classmethod 

167 def from_url(cls, url: URL, *, encoding: str = "latin1") -> Optional["BasicAuth"]: 

168 """Create BasicAuth from url.""" 

169 if not isinstance(url, URL): 

170 raise TypeError("url should be yarl.URL instance") 

171 if url.user is None: 

172 return None 

173 return cls(url.user, url.password or "", encoding=encoding) 

174 

175 def encode(self) -> str: 

176 """Encode credentials.""" 

177 creds = (f"{self.login}:{self.password}").encode(self.encoding) 

178 return "Basic %s" % base64.b64encode(creds).decode(self.encoding) 

179 

180 

181def strip_auth_from_url(url: URL) -> Tuple[URL, Optional[BasicAuth]]: 

182 auth = BasicAuth.from_url(url) 

183 if auth is None: 

184 return url, None 

185 else: 

186 return url.with_user(None), auth 

187 

188 

189def netrc_from_env() -> Optional[netrc.netrc]: 

190 """Load netrc from file. 

191 

192 Attempt to load it from the path specified by the env-var 

193 NETRC or in the default location in the user's home directory. 

194 

195 Returns None if it couldn't be found or fails to parse. 

196 """ 

197 netrc_env = os.environ.get("NETRC") 

198 

199 if netrc_env is not None: 

200 netrc_path = Path(netrc_env) 

201 else: 

202 try: 

203 home_dir = Path.home() 

204 except RuntimeError as e: # pragma: no cover 

205 # if pathlib can't resolve home, it may raise a RuntimeError 

206 client_logger.debug( 

207 "Could not resolve home directory when " 

208 "trying to look for .netrc file: %s", 

209 e, 

210 ) 

211 return None 

212 

213 netrc_path = home_dir / ( 

214 "_netrc" if platform.system() == "Windows" else ".netrc" 

215 ) 

216 

217 try: 

218 return netrc.netrc(str(netrc_path)) 

219 except netrc.NetrcParseError as e: 

220 client_logger.warning("Could not parse .netrc file: %s", e) 

221 except OSError as e: 

222 netrc_exists = False 

223 with contextlib.suppress(OSError): 

224 netrc_exists = netrc_path.is_file() 

225 # we couldn't read the file (doesn't exist, permissions, etc.) 

226 if netrc_env or netrc_exists: 

227 # only warn if the environment wanted us to load it, 

228 # or it appears like the default file does actually exist 

229 client_logger.warning("Could not read .netrc file: %s", e) 

230 

231 return None 

232 

233 

234@dataclasses.dataclass(frozen=True) 

235class ProxyInfo: 

236 proxy: URL 

237 proxy_auth: Optional[BasicAuth] 

238 

239 

240def basicauth_from_netrc(netrc_obj: Optional[netrc.netrc], host: str) -> BasicAuth: 

241 """ 

242 Return :py:class:`~aiohttp.BasicAuth` credentials for ``host`` from ``netrc_obj``. 

243 

244 :raises LookupError: if ``netrc_obj`` is :py:data:`None` or if no 

245 entry is found for the ``host``. 

246 """ 

247 if netrc_obj is None: 

248 raise LookupError("No .netrc file found") 

249 auth_from_netrc = netrc_obj.authenticators(host) 

250 

251 if auth_from_netrc is None: 

252 raise LookupError(f"No entry for {host!s} found in the `.netrc` file.") 

253 login, account, password = auth_from_netrc 

254 

255 # TODO(PY311): username = login or account 

256 # Up to python 3.10, account could be None if not specified, 

257 # and login will be empty string if not specified. From 3.11, 

258 # login and account will be empty string if not specified. 

259 username = login if (login or account is None) else account 

260 

261 # TODO(PY311): Remove this, as password will be empty string 

262 # if not specified 

263 if password is None: 

264 password = "" 

265 

266 return BasicAuth(username, password) 

267 

268 

269def proxies_from_env() -> Dict[str, ProxyInfo]: 

270 proxy_urls = { 

271 k: URL(v) 

272 for k, v in getproxies().items() 

273 if k in ("http", "https", "ws", "wss") 

274 } 

275 netrc_obj = netrc_from_env() 

276 stripped = {k: strip_auth_from_url(v) for k, v in proxy_urls.items()} 

277 ret = {} 

278 for proto, val in stripped.items(): 

279 proxy, auth = val 

280 if proxy.scheme in ("https", "wss"): 

281 client_logger.warning( 

282 "%s proxies %s are not supported, ignoring", proxy.scheme.upper(), proxy 

283 ) 

284 continue 

285 if netrc_obj and auth is None: 

286 if proxy.host is not None: 

287 try: 

288 auth = basicauth_from_netrc(netrc_obj, proxy.host) 

289 except LookupError: 

290 auth = None 

291 ret[proto] = ProxyInfo(proxy, auth) 

292 return ret 

293 

294 

295def get_env_proxy_for_url(url: URL) -> Tuple[URL, Optional[BasicAuth]]: 

296 """Get a permitted proxy for the given URL from the env.""" 

297 if url.host is not None and proxy_bypass(url.host): 

298 raise LookupError(f"Proxying is disallowed for `{url.host!r}`") 

299 

300 proxies_in_env = proxies_from_env() 

301 try: 

302 proxy_info = proxies_in_env[url.scheme] 

303 except KeyError: 

304 raise LookupError(f"No proxies found for `{url!s}` in the env") 

305 else: 

306 return proxy_info.proxy, proxy_info.proxy_auth 

307 

308 

309@dataclasses.dataclass(frozen=True) 

310class MimeType: 

311 type: str 

312 subtype: str 

313 suffix: str 

314 parameters: "MultiDictProxy[str]" 

315 

316 

317@functools.lru_cache(maxsize=56) 

318def parse_mimetype(mimetype: str) -> MimeType: 

319 """Parses a MIME type into its components. 

320 

321 mimetype is a MIME type string. 

322 

323 Returns a MimeType object. 

324 

325 Example: 

326 

327 >>> parse_mimetype('text/html; charset=utf-8') 

328 MimeType(type='text', subtype='html', suffix='', 

329 parameters={'charset': 'utf-8'}) 

330 

331 """ 

332 if not mimetype: 

333 return MimeType( 

334 type="", subtype="", suffix="", parameters=MultiDictProxy(MultiDict()) 

335 ) 

336 

337 parts = mimetype.split(";") 

338 params: MultiDict[str] = MultiDict() 

339 for item in parts[1:]: 

340 if not item: 

341 continue 

342 key, _, value = item.partition("=") 

343 params.add(key.lower().strip(), value.strip(' "')) 

344 

345 fulltype = parts[0].strip().lower() 

346 if fulltype == "*": 

347 fulltype = "*/*" 

348 

349 mtype, _, stype = fulltype.partition("/") 

350 stype, _, suffix = stype.partition("+") 

351 

352 return MimeType( 

353 type=mtype, subtype=stype, suffix=suffix, parameters=MultiDictProxy(params) 

354 ) 

355 

356 

357def guess_filename(obj: Any, default: Optional[str] = None) -> Optional[str]: 

358 name = getattr(obj, "name", None) 

359 if name and isinstance(name, str) and name[0] != "<" and name[-1] != ">": 

360 return Path(name).name 

361 return default 

362 

363 

364not_qtext_re = re.compile(r"[^\041\043-\133\135-\176]") 

365QCONTENT = {chr(i) for i in range(0x20, 0x7F)} | {"\t"} 

366 

367 

368def quoted_string(content: str) -> str: 

369 """Return 7-bit content as quoted-string. 

370 

371 Format content into a quoted-string as defined in RFC5322 for 

372 Internet Message Format. Notice that this is not the 8-bit HTTP 

373 format, but the 7-bit email format. Content must be in usascii or 

374 a ValueError is raised. 

375 """ 

376 if not (QCONTENT > set(content)): 

377 raise ValueError(f"bad content for quoted-string {content!r}") 

378 return not_qtext_re.sub(lambda x: "\\" + x.group(0), content) 

379 

380 

381def content_disposition_header( 

382 disptype: str, quote_fields: bool = True, _charset: str = "utf-8", **params: str 

383) -> str: 

384 """Sets ``Content-Disposition`` header for MIME. 

385 

386 This is the MIME payload Content-Disposition header from RFC 2183 

387 and RFC 7579 section 4.2, not the HTTP Content-Disposition from 

388 RFC 6266. 

389 

390 disptype is a disposition type: inline, attachment, form-data. 

391 Should be valid extension token (see RFC 2183) 

392 

393 quote_fields performs value quoting to 7-bit MIME headers 

394 according to RFC 7578. Set to quote_fields to False if recipient 

395 can take 8-bit file names and field values. 

396 

397 _charset specifies the charset to use when quote_fields is True. 

398 

399 params is a dict with disposition params. 

400 """ 

401 if not disptype or not (TOKEN > set(disptype)): 

402 raise ValueError("bad content disposition type {!r}" "".format(disptype)) 

403 

404 value = disptype 

405 if params: 

406 lparams = [] 

407 for key, val in params.items(): 

408 if not key or not (TOKEN > set(key)): 

409 raise ValueError( 

410 "bad content disposition parameter" " {!r}={!r}".format(key, val) 

411 ) 

412 if quote_fields: 

413 if key.lower() == "filename": 

414 qval = quote(val, "", encoding=_charset) 

415 lparams.append((key, '"%s"' % qval)) 

416 else: 

417 try: 

418 qval = quoted_string(val) 

419 except ValueError: 

420 qval = "".join( 

421 (_charset, "''", quote(val, "", encoding=_charset)) 

422 ) 

423 lparams.append((key + "*", qval)) 

424 else: 

425 lparams.append((key, '"%s"' % qval)) 

426 else: 

427 qval = val.replace("\\", "\\\\").replace('"', '\\"') 

428 lparams.append((key, '"%s"' % qval)) 

429 sparams = "; ".join("=".join(pair) for pair in lparams) 

430 value = "; ".join((value, sparams)) 

431 return value 

432 

433 

434def is_expected_content_type( 

435 response_content_type: str, expected_content_type: str 

436) -> bool: 

437 """Checks if received content type is processable as an expected one. 

438 

439 Both arguments should be given without parameters. 

440 """ 

441 if expected_content_type == "application/json": 

442 return json_re.match(response_content_type) is not None 

443 return expected_content_type in response_content_type 

444 

445 

446class _TSelf(Protocol, Generic[_T]): 

447 _cache: Dict[str, _T] 

448 

449 

450class reify(Generic[_T]): 

451 """Use as a class method decorator. 

452 

453 It operates almost exactly like 

454 the Python `@property` decorator, but it puts the result of the 

455 method it decorates into the instance dict after the first call, 

456 effectively replacing the function it decorates with an instance 

457 variable. It is, in Python parlance, a data descriptor. 

458 """ 

459 

460 def __init__(self, wrapped: Callable[..., _T]) -> None: 

461 self.wrapped = wrapped 

462 self.__doc__ = wrapped.__doc__ 

463 self.name = wrapped.__name__ 

464 

465 def __get__(self, inst: _TSelf[_T], owner: Optional[Type[Any]] = None) -> _T: 

466 try: 

467 try: 

468 return inst._cache[self.name] 

469 except KeyError: 

470 val = self.wrapped(inst) 

471 inst._cache[self.name] = val 

472 return val 

473 except AttributeError: 

474 if inst is None: 

475 return self 

476 raise 

477 

478 def __set__(self, inst: _TSelf[_T], value: _T) -> None: 

479 raise AttributeError("reified property is read-only") 

480 

481 

482reify_py = reify 

483 

484try: 

485 from ._helpers import reify as reify_c 

486 

487 if not NO_EXTENSIONS: 

488 reify = reify_c # type: ignore[misc,assignment] 

489except ImportError: 

490 pass 

491 

492_ipv4_pattern = ( 

493 r"^(?:(?:25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)\.){3}" 

494 r"(?:25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)$" 

495) 

496_ipv6_pattern = ( 

497 r"^(?:(?:(?:[A-F0-9]{1,4}:){6}|(?=(?:[A-F0-9]{0,4}:){0,6}" 

498 r"(?:[0-9]{1,3}\.){3}[0-9]{1,3}$)(([0-9A-F]{1,4}:){0,5}|:)" 

499 r"((:[0-9A-F]{1,4}){1,5}:|:)|::(?:[A-F0-9]{1,4}:){5})" 

500 r"(?:(?:25[0-5]|2[0-4][0-9]|1[0-9][0-9]|[1-9]?[0-9])\.){3}" 

501 r"(?:25[0-5]|2[0-4][0-9]|1[0-9][0-9]|[1-9]?[0-9])|(?:[A-F0-9]{1,4}:){7}" 

502 r"[A-F0-9]{1,4}|(?=(?:[A-F0-9]{0,4}:){0,7}[A-F0-9]{0,4}$)" 

503 r"(([0-9A-F]{1,4}:){1,7}|:)((:[0-9A-F]{1,4}){1,7}|:)|(?:[A-F0-9]{1,4}:){7}" 

504 r":|:(:[A-F0-9]{1,4}){7})$" 

505) 

506_ipv4_regex = re.compile(_ipv4_pattern) 

507_ipv6_regex = re.compile(_ipv6_pattern, flags=re.IGNORECASE) 

508_ipv4_regexb = re.compile(_ipv4_pattern.encode("ascii")) 

509_ipv6_regexb = re.compile(_ipv6_pattern.encode("ascii"), flags=re.IGNORECASE) 

510 

511 

512def _is_ip_address( 

513 regex: Pattern[str], regexb: Pattern[bytes], host: Optional[Union[str, bytes]] 

514) -> bool: 

515 if host is None: 

516 return False 

517 if isinstance(host, str): 

518 return bool(regex.match(host)) 

519 elif isinstance(host, (bytes, bytearray, memoryview)): 

520 return bool(regexb.match(host)) 

521 else: 

522 raise TypeError(f"{host} [{type(host)}] is not a str or bytes") 

523 

524 

525is_ipv4_address = functools.partial(_is_ip_address, _ipv4_regex, _ipv4_regexb) 

526is_ipv6_address = functools.partial(_is_ip_address, _ipv6_regex, _ipv6_regexb) 

527 

528 

529def is_ip_address(host: Optional[Union[str, bytes, bytearray, memoryview]]) -> bool: 

530 return is_ipv4_address(host) or is_ipv6_address(host) 

531 

532 

533def next_whole_second() -> datetime.datetime: 

534 """Return current time rounded up to the next whole second.""" 

535 return datetime.datetime.now(datetime.timezone.utc).replace( 

536 microsecond=0 

537 ) + datetime.timedelta(seconds=0) 

538 

539 

540_cached_current_datetime: Optional[int] = None 

541_cached_formatted_datetime = "" 

542 

543 

544def rfc822_formatted_time() -> str: 

545 global _cached_current_datetime 

546 global _cached_formatted_datetime 

547 

548 now = int(time.time()) 

549 if now != _cached_current_datetime: 

550 # Weekday and month names for HTTP date/time formatting; 

551 # always English! 

552 # Tuples are constants stored in codeobject! 

553 _weekdayname = ("Mon", "Tue", "Wed", "Thu", "Fri", "Sat", "Sun") 

554 _monthname = ( 

555 "", # Dummy so we can use 1-based month numbers 

556 "Jan", 

557 "Feb", 

558 "Mar", 

559 "Apr", 

560 "May", 

561 "Jun", 

562 "Jul", 

563 "Aug", 

564 "Sep", 

565 "Oct", 

566 "Nov", 

567 "Dec", 

568 ) 

569 

570 year, month, day, hh, mm, ss, wd, *tail = time.gmtime(now) 

571 _cached_formatted_datetime = "%s, %02d %3s %4d %02d:%02d:%02d GMT" % ( 

572 _weekdayname[wd], 

573 day, 

574 _monthname[month], 

575 year, 

576 hh, 

577 mm, 

578 ss, 

579 ) 

580 _cached_current_datetime = now 

581 return _cached_formatted_datetime 

582 

583 

584def _weakref_handle(info: "Tuple[weakref.ref[object], str]") -> None: 

585 ref, name = info 

586 ob = ref() 

587 if ob is not None: 

588 with suppress(Exception): 

589 getattr(ob, name)() 

590 

591 

592def weakref_handle( 

593 ob: object, 

594 name: str, 

595 timeout: Optional[float], 

596 loop: asyncio.AbstractEventLoop, 

597 timeout_ceil_threshold: float = 5, 

598) -> Optional[asyncio.TimerHandle]: 

599 if timeout is not None and timeout > 0: 

600 when = loop.time() + timeout 

601 if timeout >= timeout_ceil_threshold: 

602 when = ceil(when) 

603 

604 return loop.call_at(when, _weakref_handle, (weakref.ref(ob), name)) 

605 return None 

606 

607 

608def call_later( 

609 cb: Callable[[], Any], 

610 timeout: Optional[float], 

611 loop: asyncio.AbstractEventLoop, 

612 timeout_ceil_threshold: float = 5, 

613) -> Optional[asyncio.TimerHandle]: 

614 if timeout is not None and timeout > 0: 

615 when = loop.time() + timeout 

616 if timeout > timeout_ceil_threshold: 

617 when = ceil(when) 

618 return loop.call_at(when, cb) 

619 return None 

620 

621 

622class TimeoutHandle: 

623 """Timeout handle""" 

624 

625 def __init__( 

626 self, 

627 loop: asyncio.AbstractEventLoop, 

628 timeout: Optional[float], 

629 ceil_threshold: float = 5, 

630 ) -> None: 

631 self._timeout = timeout 

632 self._loop = loop 

633 self._ceil_threshold = ceil_threshold 

634 self._callbacks: List[ 

635 Tuple[Callable[..., None], Tuple[Any, ...], Dict[str, Any]] 

636 ] = [] 

637 

638 def register( 

639 self, callback: Callable[..., None], *args: Any, **kwargs: Any 

640 ) -> None: 

641 self._callbacks.append((callback, args, kwargs)) 

642 

643 def close(self) -> None: 

644 self._callbacks.clear() 

645 

646 def start(self) -> Optional[asyncio.Handle]: 

647 timeout = self._timeout 

648 if timeout is not None and timeout > 0: 

649 when = self._loop.time() + timeout 

650 if timeout >= self._ceil_threshold: 

651 when = ceil(when) 

652 return self._loop.call_at(when, self.__call__) 

653 else: 

654 return None 

655 

656 def timer(self) -> "BaseTimerContext": 

657 if self._timeout is not None and self._timeout > 0: 

658 timer = TimerContext(self._loop) 

659 self.register(timer.timeout) 

660 return timer 

661 else: 

662 return TimerNoop() 

663 

664 def __call__(self) -> None: 

665 for cb, args, kwargs in self._callbacks: 

666 with suppress(Exception): 

667 cb(*args, **kwargs) 

668 

669 self._callbacks.clear() 

670 

671 

672class BaseTimerContext(ContextManager["BaseTimerContext"]): 

673 def assert_timeout(self) -> None: 

674 """Raise TimeoutError if timeout has been exceeded.""" 

675 

676 

677class TimerNoop(BaseTimerContext): 

678 def __enter__(self) -> BaseTimerContext: 

679 return self 

680 

681 def __exit__( 

682 self, 

683 exc_type: Optional[Type[BaseException]], 

684 exc_val: Optional[BaseException], 

685 exc_tb: Optional[TracebackType], 

686 ) -> None: 

687 return 

688 

689 

690class TimerContext(BaseTimerContext): 

691 """Low resolution timeout context manager""" 

692 

693 def __init__(self, loop: asyncio.AbstractEventLoop) -> None: 

694 self._loop = loop 

695 self._tasks: List[asyncio.Task[Any]] = [] 

696 self._cancelled = False 

697 

698 def assert_timeout(self) -> None: 

699 """Raise TimeoutError if timer has already been cancelled.""" 

700 if self._cancelled: 

701 raise asyncio.TimeoutError from None 

702 

703 def __enter__(self) -> BaseTimerContext: 

704 task = asyncio.current_task(loop=self._loop) 

705 

706 if task is None: 

707 raise RuntimeError( 

708 "Timeout context manager should be used " "inside a task" 

709 ) 

710 

711 if self._cancelled: 

712 raise asyncio.TimeoutError from None 

713 

714 self._tasks.append(task) 

715 return self 

716 

717 def __exit__( 

718 self, 

719 exc_type: Optional[Type[BaseException]], 

720 exc_val: Optional[BaseException], 

721 exc_tb: Optional[TracebackType], 

722 ) -> Optional[bool]: 

723 if self._tasks: 

724 self._tasks.pop() # type: ignore[unused-awaitable] 

725 

726 if exc_type is asyncio.CancelledError and self._cancelled: 

727 raise asyncio.TimeoutError from None 

728 return None 

729 

730 def timeout(self) -> None: 

731 if not self._cancelled: 

732 for task in set(self._tasks): 

733 task.cancel() 

734 

735 self._cancelled = True 

736 

737 

738def ceil_timeout( 

739 delay: Optional[float], ceil_threshold: float = 5 

740) -> async_timeout.Timeout: 

741 if delay is None or delay <= 0: 

742 return async_timeout.timeout(None) 

743 

744 loop = asyncio.get_running_loop() 

745 now = loop.time() 

746 when = now + delay 

747 if delay > ceil_threshold: 

748 when = ceil(when) 

749 return async_timeout.timeout_at(when) 

750 

751 

752class HeadersMixin: 

753 __slots__ = ("_content_type", "_content_dict", "_stored_content_type") 

754 

755 def __init__(self) -> None: 

756 super().__init__() 

757 self._content_type: Optional[str] = None 

758 self._content_dict: Optional[Dict[str, str]] = None 

759 self._stored_content_type: Union[str, _SENTINEL] = sentinel 

760 

761 def _parse_content_type(self, raw: str) -> None: 

762 self._stored_content_type = raw 

763 if raw is None: 

764 # default value according to RFC 2616 

765 self._content_type = "application/octet-stream" 

766 self._content_dict = {} 

767 else: 

768 msg = HeaderParser().parsestr("Content-Type: " + raw) 

769 self._content_type = msg.get_content_type() 

770 params = msg.get_params(()) 

771 self._content_dict = dict(params[1:]) # First element is content type again 

772 

773 @property 

774 def content_type(self) -> str: 

775 """The value of content part for Content-Type HTTP header.""" 

776 raw = self._headers.get(hdrs.CONTENT_TYPE) # type: ignore[attr-defined] 

777 if self._stored_content_type != raw: 

778 self._parse_content_type(raw) 

779 return self._content_type # type: ignore[return-value] 

780 

781 @property 

782 def charset(self) -> Optional[str]: 

783 """The value of charset part for Content-Type HTTP header.""" 

784 raw = self._headers.get(hdrs.CONTENT_TYPE) # type: ignore[attr-defined] 

785 if self._stored_content_type != raw: 

786 self._parse_content_type(raw) 

787 return self._content_dict.get("charset") # type: ignore[union-attr] 

788 

789 @property 

790 def content_length(self) -> Optional[int]: 

791 """The value of Content-Length HTTP header.""" 

792 content_length = self._headers.get( # type: ignore[attr-defined] 

793 hdrs.CONTENT_LENGTH 

794 ) 

795 

796 if content_length is not None: 

797 return int(content_length) 

798 else: 

799 return None 

800 

801 

802def set_result(fut: "asyncio.Future[_T]", result: _T) -> None: 

803 if not fut.done(): 

804 fut.set_result(result) 

805 

806 

807def set_exception(fut: "asyncio.Future[_T]", exc: BaseException) -> None: 

808 if not fut.done(): 

809 fut.set_exception(exc) 

810 

811 

812@functools.total_ordering 

813class AppKey(Generic[_T]): 

814 """Keys for static typing support in Application.""" 

815 

816 __slots__ = ("_name", "_t", "__orig_class__") 

817 

818 # This may be set by Python when instantiating with a generic type. We need to 

819 # support this, in order to support types that are not concrete classes, 

820 # like Iterable, which can't be passed as the second parameter to __init__. 

821 __orig_class__: Type[object] 

822 

823 def __init__(self, name: str, t: Optional[Type[_T]] = None): 

824 # Prefix with module name to help deduplicate key names. 

825 frame = inspect.currentframe() 

826 while frame: 

827 if frame.f_code.co_name == "<module>": 

828 module: str = frame.f_globals["__name__"] 

829 break 

830 frame = frame.f_back 

831 else: 

832 raise RuntimeError("Failed to get module name.") 

833 

834 # https://github.com/python/mypy/issues/14209 

835 self._name = module + "." + name # type: ignore[possibly-undefined]