Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/aiohttp/abc.py: 87%
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
1import asyncio
2import logging
3import socket
4from abc import ABC, abstractmethod
5from collections.abc import Awaitable, Callable, Generator, Iterable, Sequence, Sized
6from http.cookies import BaseCookie, Morsel, SimpleCookie
7from types import MappingProxyType
8from typing import TYPE_CHECKING, Any, TypedDict
10from multidict import CIMultiDict
11from yarl import URL
13from ._cookie_helpers import parse_set_cookie_headers
14from .typedefs import LooseCookies
16if TYPE_CHECKING:
17 from .web_app import Application
18 from .web_exceptions import HTTPException
19 from .web_request import BaseRequest, Request
20 from .web_response import StreamResponse
21else:
22 BaseRequest = Request = Application = StreamResponse = Any
23 HTTPException = Any
26class AbstractRouter(ABC):
27 def __init__(self) -> None:
28 self._frozen = False
30 def post_init(self, app: Application) -> None:
31 """Post init stage.
33 Not an abstract method for sake of backward compatibility,
34 but if the router wants to be aware of the application
35 it can override this.
36 """
38 @property
39 def frozen(self) -> bool:
40 return self._frozen
42 def freeze(self) -> None:
43 """Freeze router."""
44 self._frozen = True
46 @abstractmethod
47 async def resolve(self, request: Request) -> "AbstractMatchInfo":
48 """Return MATCH_INFO for given request"""
51class AbstractMatchInfo(ABC):
53 __slots__ = ()
55 @property # pragma: no branch
56 @abstractmethod
57 def handler(self) -> Callable[[Request], Awaitable[StreamResponse]]:
58 """Execute matched request handler"""
60 @property
61 @abstractmethod
62 def expect_handler(
63 self,
64 ) -> Callable[[Request], Awaitable[StreamResponse | None]]:
65 """Expect handler for 100-continue processing"""
67 @property # pragma: no branch
68 @abstractmethod
69 def http_exception(self) -> HTTPException | None:
70 """HTTPException instance raised on router's resolving, or None"""
72 @abstractmethod # pragma: no branch
73 def get_info(self) -> dict[str, Any]:
74 """Return a dict with additional info useful for introspection"""
76 @property # pragma: no branch
77 @abstractmethod
78 def apps(self) -> tuple[Application, ...]:
79 """Stack of nested applications.
81 Top level application is left-most element.
83 """
85 @abstractmethod
86 def add_app(self, app: Application) -> None:
87 """Add application to the nested apps stack."""
89 @abstractmethod
90 def freeze(self) -> None:
91 """Freeze the match info.
93 The method is called after route resolution.
95 After the call .add_app() is forbidden.
97 """
100class AbstractView(ABC):
101 """Abstract class based view."""
103 def __init__(self, request: Request) -> None:
104 self._request = request
106 @property
107 def request(self) -> Request:
108 """Request instance."""
109 return self._request
111 @abstractmethod
112 def __await__(self) -> Generator[None, None, StreamResponse]:
113 """Execute the view handler."""
116class ResolveResult(TypedDict):
117 """Resolve result.
119 This is the result returned from an AbstractResolver's
120 resolve method.
122 :param hostname: The hostname that was provided.
123 :param host: The IP address that was resolved.
124 :param port: The port that was resolved.
125 :param family: The address family that was resolved.
126 :param proto: The protocol that was resolved.
127 :param flags: The flags that were resolved.
128 """
130 hostname: str
131 host: str
132 port: int
133 family: int
134 proto: int
135 flags: int
138class AbstractResolver(ABC):
139 """Abstract DNS resolver."""
141 @abstractmethod
142 async def resolve(
143 self, host: str, port: int = 0, family: socket.AddressFamily = socket.AF_INET
144 ) -> list[ResolveResult]:
145 """Return IP address for given hostname"""
147 @abstractmethod
148 async def close(self) -> None:
149 """Release resolver"""
152if TYPE_CHECKING:
153 IterableBase = Iterable[Morsel[str]]
154else:
155 IterableBase = Iterable
158ClearCookiePredicate = Callable[["Morsel[str]"], bool]
161class AbstractCookieJar(Sized, IterableBase):
162 """Abstract Cookie Jar."""
164 def __init__(self, *, loop: asyncio.AbstractEventLoop | None = None) -> None:
165 self._loop = loop or asyncio.get_running_loop()
167 @property
168 @abstractmethod
169 def unsafe(self) -> bool:
170 """Return True if cookies can be used with IP addresses."""
172 @property
173 @abstractmethod
174 def quote_cookie(self) -> bool:
175 """Return True if cookies should be quoted."""
177 @property
178 def treat_as_secure_origin(self) -> frozenset[URL]:
179 """Return origins considered secure even over cleartext connections."""
180 return frozenset()
182 @property
183 @abstractmethod
184 def cookies(self) -> MappingProxyType[tuple[str, str], SimpleCookie]:
185 """Return the cookies stored in this jar."""
187 @property
188 @abstractmethod
189 def host_only_cookies(self) -> frozenset[tuple[str, str, str]]:
190 """Return the host-only cookies stored in this jar."""
192 @abstractmethod
193 def clear(self, predicate: ClearCookiePredicate | None = None) -> None:
194 """Clear all cookies if no predicate is passed."""
196 @abstractmethod
197 def clear_domain(self, domain: str) -> None:
198 """Clear all cookies for domain and all subdomains."""
200 @abstractmethod
201 def update_cookies(self, cookies: LooseCookies, response_url: URL = URL()) -> None:
202 """Update cookies."""
204 def update_cookies_from_headers(
205 self, headers: Sequence[str], response_url: URL
206 ) -> None:
207 """Update cookies from raw Set-Cookie headers."""
208 if headers and (cookies_to_update := parse_set_cookie_headers(headers)):
209 self.update_cookies(cookies_to_update, response_url)
211 @abstractmethod
212 def filter_cookies(self, request_url: URL) -> "BaseCookie[str]":
213 """Return the jar's cookies filtered by their attributes."""
216class AbstractStreamWriter(ABC):
217 """Abstract stream writer."""
219 buffer_size: int = 0
220 output_size: int = 0
221 length: int | None = 0
222 # Called with each accepted body chunk's byte count (before any
223 # transport-level transformation such as compression or chunked
224 # framing) and the writer's total output_size after the chunk.
225 # Assigned by the client request machinery for upload progress
226 # tracking; write()/write_eof() implementations should invoke it
227 # for every body chunk they accept.
228 on_body_write: Callable[[int, int], None] | None = None
230 @property
231 def transport(self) -> asyncio.WriteTransport | None:
232 """The transport this writer writes to, if any.
234 Used by upload progress tracking to observe the unsent buffer;
235 writers without one report progress only on completion.
236 """
237 return None
239 @abstractmethod
240 async def write(self, chunk: bytes | bytearray | memoryview) -> None:
241 """Write chunk into stream."""
243 @abstractmethod
244 async def write_eof(self, chunk: bytes = b"") -> None:
245 """Write last chunk."""
247 @abstractmethod
248 async def drain(self) -> None:
249 """Flush the write buffer."""
251 @abstractmethod
252 def enable_compression(
253 self, encoding: str = "deflate", strategy: int | None = None
254 ) -> None:
255 """Enable HTTP body compression"""
257 @abstractmethod
258 def enable_chunking(self) -> None:
259 """Enable HTTP chunked mode"""
261 @abstractmethod
262 async def write_headers(
263 self, status_line: str, headers: "CIMultiDict[str]"
264 ) -> None:
265 """Write HTTP headers"""
267 def send_headers(self) -> None:
268 """Force sending buffered headers if not already sent.
270 Required only if write_headers() buffers headers instead of sending immediately.
271 For backwards compatibility, this method does nothing by default.
272 """
275class AbstractAccessLogger(ABC):
276 """Abstract writer to access log."""
278 __slots__ = ("logger", "log_format")
280 def __init__(self, logger: logging.Logger, log_format: str) -> None:
281 self.logger = logger
282 self.log_format = log_format
284 @abstractmethod
285 def log(self, request: BaseRequest, response: StreamResponse, time: float) -> None:
286 """Emit log to logger."""
288 @property
289 def enabled(self) -> bool:
290 """Check if logger is enabled."""
291 return True