Coverage for /pythoncovmergedfiles/medio/medio/src/aiohttp/aiohttp/abc.py: 88%
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"""
152ClearCookiePredicate = Callable[[Morsel[str]], bool]
155class AbstractCookieJar(Sized, Iterable[Morsel[str]]):
156 """Abstract Cookie Jar."""
158 @property
159 @abstractmethod
160 def unsafe(self) -> bool:
161 """Return True if cookies can be used with IP addresses."""
163 @property
164 @abstractmethod
165 def quote_cookie(self) -> bool:
166 """Return True if cookies should be quoted."""
168 @property
169 def treat_as_secure_origin(self) -> frozenset[URL]:
170 """Return origins considered secure even over cleartext connections."""
171 return frozenset()
173 @property
174 @abstractmethod
175 def cookies(self) -> MappingProxyType[tuple[str, str], SimpleCookie]:
176 """Return the cookies stored in this jar."""
178 @property
179 @abstractmethod
180 def host_only_cookies(self) -> frozenset[tuple[str, str, str]]:
181 """Return the host-only cookies stored in this jar."""
183 @abstractmethod
184 def clear(self, predicate: ClearCookiePredicate | None = None) -> None:
185 """Clear all cookies if no predicate is passed."""
187 @abstractmethod
188 def clear_domain(self, domain: str) -> None:
189 """Clear all cookies for domain and all subdomains."""
191 @abstractmethod
192 def update_cookies(self, cookies: LooseCookies, response_url: URL = URL()) -> None:
193 """Update cookies."""
195 def update_cookies_from_headers(
196 self, headers: Sequence[str], response_url: URL
197 ) -> None:
198 """Update cookies from raw Set-Cookie headers."""
199 if headers and (cookies_to_update := parse_set_cookie_headers(headers)):
200 self.update_cookies(cookies_to_update, response_url)
202 @abstractmethod
203 def filter_cookies(self, request_url: URL) -> BaseCookie[str]:
204 """Return the jar's cookies filtered by their attributes."""
207class AbstractStreamWriter(ABC):
208 """Abstract stream writer."""
210 buffer_size: int = 0
211 output_size: int = 0
212 length: int | None = 0
213 # Called with each accepted body chunk's byte count (before any
214 # transport-level transformation such as compression or chunked
215 # framing) and the writer's total output_size after the chunk.
216 # Assigned by the client request machinery for upload progress
217 # tracking; write()/write_eof() implementations should invoke it
218 # for every body chunk they accept.
219 on_body_write: Callable[[int, int], None] | None = None
221 @property
222 def transport(self) -> asyncio.WriteTransport | None:
223 """The transport this writer writes to, if any.
225 Used by upload progress tracking to observe the unsent buffer;
226 writers without one report progress only on completion.
227 """
228 return None
230 @abstractmethod
231 async def write(
232 self, chunk: "bytes | bytearray | memoryview[int] | memoryview[bytes]"
233 ) -> None:
234 """Write chunk into stream."""
236 @abstractmethod
237 async def write_eof(self, chunk: bytes = b"") -> None:
238 """Write last chunk."""
240 @abstractmethod
241 async def drain(self) -> None:
242 """Flush the write buffer."""
244 @abstractmethod
245 def enable_compression(
246 self, encoding: str = "deflate", strategy: int | None = None
247 ) -> None:
248 """Enable HTTP body compression"""
250 @abstractmethod
251 def enable_chunking(self) -> None:
252 """Enable HTTP chunked mode"""
254 @abstractmethod
255 async def write_headers(self, status_line: str, headers: CIMultiDict[str]) -> None:
256 """Write HTTP headers"""
258 def send_headers(self) -> None:
259 """Force sending buffered headers if not already sent.
261 Required only if write_headers() buffers headers instead of sending immediately.
262 For backwards compatibility, this method does nothing by default.
263 """
266class AbstractAccessLogger(ABC):
267 """Abstract writer to access log."""
269 __slots__ = ("logger", "log_format")
271 def __init__(self, logger: logging.Logger, log_format: str) -> None:
272 self.logger = logger
273 self.log_format = log_format
275 @abstractmethod
276 def log(self, request: BaseRequest, response: StreamResponse, time: float) -> None:
277 """Emit log to logger."""
279 @property
280 def enabled(self) -> bool:
281 """Check if logger is enabled."""
282 return True
285class AbstractAsyncAccessLogger(ABC):
286 """Abstract asynchronous writer to access log."""
288 __slots__ = ()
290 @abstractmethod
291 async def log(
292 self, request: BaseRequest, response: StreamResponse, request_start: float
293 ) -> None:
294 """Emit log to logger."""
296 @property
297 def enabled(self) -> bool:
298 """Check if logger is enabled."""
299 return True