codekingpro/portable-devtools

1import asyncio
2import logging
3import socket
4from abc import ABC, abstractmethod
5from collections.abc import Sized
6from http.cookies import BaseCookie, Morsel
7from typing import (
8    TYPE_CHECKING,
9    Any,
10    Awaitable,
11    Callable,
12    Dict,
13    Generator,
14    Iterable,
15    List,
16    Optional,
17    Sequence,
18    Tuple,
19    TypedDict,
20    Union,
21)
22
23from multidict import CIMultiDict
24from yarl import URL
25
26from ._cookie_helpers import parse_set_cookie_headers
27from .typedefs import LooseCookies
28
29if TYPE_CHECKING:
30    from .web_app import Application
31    from .web_exceptions import HTTPException
32    from .web_request import BaseRequest, Request
33    from .web_response import StreamResponse
34else:
35    BaseRequest = Request = Application = StreamResponse = None
36    HTTPException = None
37
38
39class AbstractRouter(ABC):
40    def __init__(self) -> None:
41        self._frozen = False
42
43    def post_init(self, app: Application) -> None:
44        """Post init stage.
45
46        Not an abstract method for sake of backward compatibility,
47        but if the router wants to be aware of the application
48        it can override this.
49        """
50
51    @property
52    def frozen(self) -> bool:
53        return self._frozen
54
55    def freeze(self) -> None:
56        """Freeze router."""
57        self._frozen = True
58
59    @abstractmethod
60    async def resolve(self, request: Request) -> "AbstractMatchInfo":
61        """Return MATCH_INFO for given request"""
62
63
64class AbstractMatchInfo(ABC):
65
66    __slots__ = ()
67
68    @property  # pragma: no branch
69    @abstractmethod
70    def handler(self) -> Callable[[Request], Awaitable[StreamResponse]]:
71        """Execute matched request handler"""
72
73    @property
74    @abstractmethod
75    def expect_handler(
76        self,
77    ) -> Callable[[Request], Awaitable[Optional[StreamResponse]]]:
78        """Expect handler for 100-continue processing"""
79
80    @property  # pragma: no branch
81    @abstractmethod
82    def http_exception(self) -> Optional[HTTPException]:
83        """HTTPException instance raised on router's resolving, or None"""
84
85    @abstractmethod  # pragma: no branch
86    def get_info(self) -> Dict[str, Any]:
87        """Return a dict with additional info useful for introspection"""
88
89    @property  # pragma: no branch
90    @abstractmethod
91    def apps(self) -> Tuple[Application, ...]:
92        """Stack of nested applications.
93
94        Top level application is left-most element.
95
96        """
97
98    @abstractmethod
99    def add_app(self, app: Application) -> None:
100        """Add application to the nested apps stack."""
101
102    @abstractmethod
103    def freeze(self) -> None:
104        """Freeze the match info.
105
106        The method is called after route resolution.
107
108        After the call .add_app() is forbidden.
109
110        """
111
112
113class AbstractView(ABC):
114    """Abstract class based view."""
115
116    def __init__(self, request: Request) -> None:
117        self._request = request
118
119    @property
120    def request(self) -> Request:
121        """Request instance."""
122        return self._request
123
124    @abstractmethod
125    def __await__(self) -> Generator[None, None, StreamResponse]:
126        """Execute the view handler."""
127
128
129class ResolveResult(TypedDict):
130    """Resolve result.
131
132    This is the result returned from an AbstractResolver's
133    resolve method.
134
135    :param hostname: The hostname that was provided.
136    :param host: The IP address that was resolved.
137    :param port: The port that was resolved.
138    :param family: The address family that was resolved.
139    :param proto: The protocol that was resolved.
140    :param flags: The flags that were resolved.
141    """
142
143    hostname: str
144    host: str
145    port: int
146    family: int
147    proto: int
148    flags: int
149
150
151class AbstractResolver(ABC):
152    """Abstract DNS resolver."""
153
154    @abstractmethod
155    async def resolve(
156        self, host: str, port: int = 0, family: socket.AddressFamily = socket.AF_INET
157    ) -> List[ResolveResult]:
158        """Return IP address for given hostname"""
159
160    @abstractmethod
161    async def close(self) -> None:
162        """Release resolver"""
163
164
165if TYPE_CHECKING:
166    IterableBase = Iterable[Morsel[str]]
167else:
168    IterableBase = Iterable
169
170
171ClearCookiePredicate = Callable[["Morsel[str]"], bool]
172
173
174class AbstractCookieJar(Sized, IterableBase):
175    """Abstract Cookie Jar."""
176
177    def __init__(self, *, loop: Optional[asyncio.AbstractEventLoop] = None) -> None:
178        self._loop = loop or asyncio.get_running_loop()
179
180    @property
181    @abstractmethod
182    def quote_cookie(self) -> bool:
183        """Return True if cookies should be quoted."""
184
185    @abstractmethod
186    def clear(self, predicate: Optional[ClearCookiePredicate] = None) -> None:
187        """Clear all cookies if no predicate is passed."""
188
189    @abstractmethod
190    def clear_domain(self, domain: str) -> None:
191        """Clear all cookies for domain and all subdomains."""
192
193    @abstractmethod
194    def update_cookies(self, cookies: LooseCookies, response_url: URL = URL()) -> None:
195        """Update cookies."""
196
197    def update_cookies_from_headers(
198        self, headers: Sequence[str], response_url: URL
199    ) -> None:
200        """Update cookies from raw Set-Cookie headers."""
201        if headers and (cookies_to_update := parse_set_cookie_headers(headers)):
202            self.update_cookies(cookies_to_update, response_url)
203
204    @abstractmethod
205    def filter_cookies(self, request_url: URL) -> "BaseCookie[str]":
206        """Return the jar's cookies filtered by their attributes."""
207
208
209class AbstractStreamWriter(ABC):
210    """Abstract stream writer."""
211
212    buffer_size: int = 0
213    output_size: int = 0
214    length: Optional[int] = 0
215
216    @abstractmethod
217    async def write(self, chunk: Union[bytes, bytearray, memoryview]) -> None:
218        """Write chunk into stream."""
219
220    @abstractmethod
221    async def write_eof(self, chunk: bytes = b"") -> None:
222        """Write last chunk."""
223
224    @abstractmethod
225    async def drain(self) -> None:
226        """Flush the write buffer."""
227
228    @abstractmethod
229    def enable_compression(
230        self, encoding: str = "deflate", strategy: Optional[int] = None
231    ) -> None:
232        """Enable HTTP body compression"""
233
234    @abstractmethod
235    def enable_chunking(self) -> None:
236        """Enable HTTP chunked mode"""
237
238    @abstractmethod
239    async def write_headers(
240        self, status_line: str, headers: "CIMultiDict[str]"
241    ) -> None:
242        """Write HTTP headers"""
243
244    def send_headers(self) -> None:
245        """Force sending buffered headers if not already sent.
246
247        Required only if write_headers() buffers headers instead of sending immediately.
248        For backwards compatibility, this method does nothing by default.
249        """
250
251
252class AbstractAccessLogger(ABC):
253    """Abstract writer to access log."""
254
255    __slots__ = ("logger", "log_format")
256
257    def __init__(self, logger: logging.Logger, log_format: str) -> None:
258        self.logger = logger
259        self.log_format = log_format
260
261    @abstractmethod
262    def log(self, request: BaseRequest, response: StreamResponse, time: float) -> None:
263        """Emit log to logger."""
264
265    @property
266    def enabled(self) -> bool:
267        """Check if logger is enabled."""
268        return True
269