Path: blob/master/ invest-robot-contest_TinkoffBotTwitch-main/venv/lib/python3.8/site-packages/aiohttp/abc.py
7762 views
import asyncio1import logging2from abc import ABC, abstractmethod3from collections.abc import Sized4from http.cookies import BaseCookie, Morsel5from typing import (6TYPE_CHECKING,7Any,8Awaitable,9Callable,10Dict,11Generator,12Iterable,13List,14Optional,15Tuple,16)1718from multidict import CIMultiDict19from yarl import URL2021from .helpers import get_running_loop22from .typedefs import LooseCookies2324if TYPE_CHECKING: # pragma: no cover25from .web_app import Application26from .web_exceptions import HTTPException27from .web_request import BaseRequest, Request28from .web_response import StreamResponse29else:30BaseRequest = Request = Application = StreamResponse = None31HTTPException = None323334class AbstractRouter(ABC):35def __init__(self) -> None:36self._frozen = False3738def post_init(self, app: Application) -> None:39"""Post init stage.4041Not an abstract method for sake of backward compatibility,42but if the router wants to be aware of the application43it can override this.44"""4546@property47def frozen(self) -> bool:48return self._frozen4950def freeze(self) -> None:51"""Freeze router."""52self._frozen = True5354@abstractmethod55async def resolve(self, request: Request) -> "AbstractMatchInfo":56"""Return MATCH_INFO for given request"""575859class AbstractMatchInfo(ABC):60@property # pragma: no branch61@abstractmethod62def handler(self) -> Callable[[Request], Awaitable[StreamResponse]]:63"""Execute matched request handler"""6465@property66@abstractmethod67def expect_handler(self) -> Callable[[Request], Awaitable[None]]:68"""Expect handler for 100-continue processing"""6970@property # pragma: no branch71@abstractmethod72def http_exception(self) -> Optional[HTTPException]:73"""HTTPException instance raised on router's resolving, or None"""7475@abstractmethod # pragma: no branch76def get_info(self) -> Dict[str, Any]:77"""Return a dict with additional info useful for introspection"""7879@property # pragma: no branch80@abstractmethod81def apps(self) -> Tuple[Application, ...]:82"""Stack of nested applications.8384Top level application is left-most element.8586"""8788@abstractmethod89def add_app(self, app: Application) -> None:90"""Add application to the nested apps stack."""9192@abstractmethod93def freeze(self) -> None:94"""Freeze the match info.9596The method is called after route resolution.9798After the call .add_app() is forbidden.99100"""101102103class AbstractView(ABC):104"""Abstract class based view."""105106def __init__(self, request: Request) -> None:107self._request = request108109@property110def request(self) -> Request:111"""Request instance."""112return self._request113114@abstractmethod115def __await__(self) -> Generator[Any, None, StreamResponse]:116"""Execute the view handler."""117118119class AbstractResolver(ABC):120"""Abstract DNS resolver."""121122@abstractmethod123async def resolve(self, host: str, port: int, family: int) -> List[Dict[str, Any]]:124"""Return IP address for given hostname"""125126@abstractmethod127async def close(self) -> None:128"""Release resolver"""129130131if TYPE_CHECKING: # pragma: no cover132IterableBase = Iterable[Morsel[str]]133else:134IterableBase = Iterable135136137ClearCookiePredicate = Callable[["Morsel[str]"], bool]138139140class AbstractCookieJar(Sized, IterableBase):141"""Abstract Cookie Jar."""142143def __init__(self, *, loop: Optional[asyncio.AbstractEventLoop] = None) -> None:144self._loop = get_running_loop(loop)145146@abstractmethod147def clear(self, predicate: Optional[ClearCookiePredicate] = None) -> None:148"""Clear all cookies if no predicate is passed."""149150@abstractmethod151def clear_domain(self, domain: str) -> None:152"""Clear all cookies for domain and all subdomains."""153154@abstractmethod155def update_cookies(self, cookies: LooseCookies, response_url: URL = URL()) -> None:156"""Update cookies."""157158@abstractmethod159def filter_cookies(self, request_url: URL) -> "BaseCookie[str]":160"""Return the jar's cookies filtered by their attributes."""161162163class AbstractStreamWriter(ABC):164"""Abstract stream writer."""165166buffer_size = 0167output_size = 0168length = 0 # type: Optional[int]169170@abstractmethod171async def write(self, chunk: bytes) -> None:172"""Write chunk into stream."""173174@abstractmethod175async def write_eof(self, chunk: bytes = b"") -> None:176"""Write last chunk."""177178@abstractmethod179async def drain(self) -> None:180"""Flush the write buffer."""181182@abstractmethod183def enable_compression(self, encoding: str = "deflate") -> None:184"""Enable HTTP body compression"""185186@abstractmethod187def enable_chunking(self) -> None:188"""Enable HTTP chunked mode"""189190@abstractmethod191async def write_headers(192self, status_line: str, headers: "CIMultiDict[str]"193) -> None:194"""Write HTTP headers"""195196197class AbstractAccessLogger(ABC):198"""Abstract writer to access log."""199200def __init__(self, logger: logging.Logger, log_format: str) -> None:201self.logger = logger202self.log_format = log_format203204@abstractmethod205def log(self, request: BaseRequest, response: StreamResponse, time: float) -> None:206"""Emit log to logger."""207208209