Book a Demo!
CoCalc Logo Icon
StoreFeaturesDocsShareSupportNewsAboutPoliciesSign UpSign In
wiseplat
GitHub Repository: wiseplat/python-code
Path: blob/master/ invest-robot-contest_TinkoffBotTwitch-main/venv/lib/python3.8/site-packages/aiohttp/abc.py
7762 views
1
import asyncio
2
import logging
3
from abc import ABC, abstractmethod
4
from collections.abc import Sized
5
from http.cookies import BaseCookie, Morsel
6
from typing import (
7
TYPE_CHECKING,
8
Any,
9
Awaitable,
10
Callable,
11
Dict,
12
Generator,
13
Iterable,
14
List,
15
Optional,
16
Tuple,
17
)
18
19
from multidict import CIMultiDict
20
from yarl import URL
21
22
from .helpers import get_running_loop
23
from .typedefs import LooseCookies
24
25
if TYPE_CHECKING: # pragma: no cover
26
from .web_app import Application
27
from .web_exceptions import HTTPException
28
from .web_request import BaseRequest, Request
29
from .web_response import StreamResponse
30
else:
31
BaseRequest = Request = Application = StreamResponse = None
32
HTTPException = None
33
34
35
class AbstractRouter(ABC):
36
def __init__(self) -> None:
37
self._frozen = False
38
39
def post_init(self, app: Application) -> None:
40
"""Post init stage.
41
42
Not an abstract method for sake of backward compatibility,
43
but if the router wants to be aware of the application
44
it can override this.
45
"""
46
47
@property
48
def frozen(self) -> bool:
49
return self._frozen
50
51
def freeze(self) -> None:
52
"""Freeze router."""
53
self._frozen = True
54
55
@abstractmethod
56
async def resolve(self, request: Request) -> "AbstractMatchInfo":
57
"""Return MATCH_INFO for given request"""
58
59
60
class AbstractMatchInfo(ABC):
61
@property # pragma: no branch
62
@abstractmethod
63
def handler(self) -> Callable[[Request], Awaitable[StreamResponse]]:
64
"""Execute matched request handler"""
65
66
@property
67
@abstractmethod
68
def expect_handler(self) -> Callable[[Request], Awaitable[None]]:
69
"""Expect handler for 100-continue processing"""
70
71
@property # pragma: no branch
72
@abstractmethod
73
def http_exception(self) -> Optional[HTTPException]:
74
"""HTTPException instance raised on router's resolving, or None"""
75
76
@abstractmethod # pragma: no branch
77
def get_info(self) -> Dict[str, Any]:
78
"""Return a dict with additional info useful for introspection"""
79
80
@property # pragma: no branch
81
@abstractmethod
82
def apps(self) -> Tuple[Application, ...]:
83
"""Stack of nested applications.
84
85
Top level application is left-most element.
86
87
"""
88
89
@abstractmethod
90
def add_app(self, app: Application) -> None:
91
"""Add application to the nested apps stack."""
92
93
@abstractmethod
94
def freeze(self) -> None:
95
"""Freeze the match info.
96
97
The method is called after route resolution.
98
99
After the call .add_app() is forbidden.
100
101
"""
102
103
104
class AbstractView(ABC):
105
"""Abstract class based view."""
106
107
def __init__(self, request: Request) -> None:
108
self._request = request
109
110
@property
111
def request(self) -> Request:
112
"""Request instance."""
113
return self._request
114
115
@abstractmethod
116
def __await__(self) -> Generator[Any, None, StreamResponse]:
117
"""Execute the view handler."""
118
119
120
class AbstractResolver(ABC):
121
"""Abstract DNS resolver."""
122
123
@abstractmethod
124
async def resolve(self, host: str, port: int, family: int) -> List[Dict[str, Any]]:
125
"""Return IP address for given hostname"""
126
127
@abstractmethod
128
async def close(self) -> None:
129
"""Release resolver"""
130
131
132
if TYPE_CHECKING: # pragma: no cover
133
IterableBase = Iterable[Morsel[str]]
134
else:
135
IterableBase = Iterable
136
137
138
ClearCookiePredicate = Callable[["Morsel[str]"], bool]
139
140
141
class AbstractCookieJar(Sized, IterableBase):
142
"""Abstract Cookie Jar."""
143
144
def __init__(self, *, loop: Optional[asyncio.AbstractEventLoop] = None) -> None:
145
self._loop = get_running_loop(loop)
146
147
@abstractmethod
148
def clear(self, predicate: Optional[ClearCookiePredicate] = None) -> None:
149
"""Clear all cookies if no predicate is passed."""
150
151
@abstractmethod
152
def clear_domain(self, domain: str) -> None:
153
"""Clear all cookies for domain and all subdomains."""
154
155
@abstractmethod
156
def update_cookies(self, cookies: LooseCookies, response_url: URL = URL()) -> None:
157
"""Update cookies."""
158
159
@abstractmethod
160
def filter_cookies(self, request_url: URL) -> "BaseCookie[str]":
161
"""Return the jar's cookies filtered by their attributes."""
162
163
164
class AbstractStreamWriter(ABC):
165
"""Abstract stream writer."""
166
167
buffer_size = 0
168
output_size = 0
169
length = 0 # type: Optional[int]
170
171
@abstractmethod
172
async def write(self, chunk: bytes) -> None:
173
"""Write chunk into stream."""
174
175
@abstractmethod
176
async def write_eof(self, chunk: bytes = b"") -> None:
177
"""Write last chunk."""
178
179
@abstractmethod
180
async def drain(self) -> None:
181
"""Flush the write buffer."""
182
183
@abstractmethod
184
def enable_compression(self, encoding: str = "deflate") -> None:
185
"""Enable HTTP body compression"""
186
187
@abstractmethod
188
def enable_chunking(self) -> None:
189
"""Enable HTTP chunked mode"""
190
191
@abstractmethod
192
async def write_headers(
193
self, status_line: str, headers: "CIMultiDict[str]"
194
) -> None:
195
"""Write HTTP headers"""
196
197
198
class AbstractAccessLogger(ABC):
199
"""Abstract writer to access log."""
200
201
def __init__(self, logger: logging.Logger, log_format: str) -> None:
202
self.logger = logger
203
self.log_format = log_format
204
205
@abstractmethod
206
def log(self, request: BaseRequest, response: StreamResponse, time: float) -> None:
207
"""Emit log to logger."""
208
209