Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/aiohttp/abc.py: 88%

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 

9 

10from multidict import CIMultiDict 

11from yarl import URL 

12 

13from ._cookie_helpers import parse_set_cookie_headers 

14from .typedefs import LooseCookies 

15 

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 

24 

25 

26class AbstractRouter(ABC): 

27 def __init__(self) -> None: 

28 self._frozen = False 

29 

30 def post_init(self, app: Application) -> None: 

31 """Post init stage. 

32 

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 """ 

37 

38 @property 

39 def frozen(self) -> bool: 

40 return self._frozen 

41 

42 def freeze(self) -> None: 

43 """Freeze router.""" 

44 self._frozen = True 

45 

46 @abstractmethod 

47 async def resolve(self, request: Request) -> "AbstractMatchInfo": 

48 """Return MATCH_INFO for given request""" 

49 

50 

51class AbstractMatchInfo(ABC): 

52 

53 __slots__ = () 

54 

55 @property # pragma: no branch 

56 @abstractmethod 

57 def handler(self) -> Callable[[Request], Awaitable[StreamResponse]]: 

58 """Execute matched request handler""" 

59 

60 @property 

61 @abstractmethod 

62 def expect_handler( 

63 self, 

64 ) -> Callable[[Request], Awaitable[StreamResponse | None]]: 

65 """Expect handler for 100-continue processing""" 

66 

67 @property # pragma: no branch 

68 @abstractmethod 

69 def http_exception(self) -> HTTPException | None: 

70 """HTTPException instance raised on router's resolving, or None""" 

71 

72 @abstractmethod # pragma: no branch 

73 def get_info(self) -> dict[str, Any]: 

74 """Return a dict with additional info useful for introspection""" 

75 

76 @property # pragma: no branch 

77 @abstractmethod 

78 def apps(self) -> tuple[Application, ...]: 

79 """Stack of nested applications. 

80 

81 Top level application is left-most element. 

82 

83 """ 

84 

85 @abstractmethod 

86 def add_app(self, app: Application) -> None: 

87 """Add application to the nested apps stack.""" 

88 

89 @abstractmethod 

90 def freeze(self) -> None: 

91 """Freeze the match info. 

92 

93 The method is called after route resolution. 

94 

95 After the call .add_app() is forbidden. 

96 

97 """ 

98 

99 

100class AbstractView(ABC): 

101 """Abstract class based view.""" 

102 

103 def __init__(self, request: Request) -> None: 

104 self._request = request 

105 

106 @property 

107 def request(self) -> Request: 

108 """Request instance.""" 

109 return self._request 

110 

111 @abstractmethod 

112 def __await__(self) -> Generator[None, None, StreamResponse]: 

113 """Execute the view handler.""" 

114 

115 

116class ResolveResult(TypedDict): 

117 """Resolve result. 

118 

119 This is the result returned from an AbstractResolver's 

120 resolve method. 

121 

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 """ 

129 

130 hostname: str 

131 host: str 

132 port: int 

133 family: int 

134 proto: int 

135 flags: int 

136 

137 

138class AbstractResolver(ABC): 

139 """Abstract DNS resolver.""" 

140 

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""" 

146 

147 @abstractmethod 

148 async def close(self) -> None: 

149 """Release resolver""" 

150 

151 

152if TYPE_CHECKING: 

153 IterableBase = Iterable[Morsel[str]] 

154else: 

155 IterableBase = Iterable 

156 

157 

158ClearCookiePredicate = Callable[["Morsel[str]"], bool] 

159 

160 

161class AbstractCookieJar(Sized, IterableBase): 

162 """Abstract Cookie Jar.""" 

163 

164 def __init__(self, *, loop: asyncio.AbstractEventLoop | None = None) -> None: 

165 self._loop = loop or asyncio.get_running_loop() 

166 

167 @property 

168 @abstractmethod 

169 def unsafe(self) -> bool: 

170 """Return True if cookies can be used with IP addresses.""" 

171 

172 @property 

173 @abstractmethod 

174 def quote_cookie(self) -> bool: 

175 """Return True if cookies should be quoted.""" 

176 

177 @property 

178 @abstractmethod 

179 def cookies(self) -> MappingProxyType[tuple[str, str], SimpleCookie]: 

180 """Return the cookies stored in this jar.""" 

181 

182 @property 

183 @abstractmethod 

184 def host_only_cookies(self) -> frozenset[tuple[str, str]]: 

185 """Return the host-only cookies stored in this jar.""" 

186 

187 @abstractmethod 

188 def clear(self, predicate: ClearCookiePredicate | None = None) -> None: 

189 """Clear all cookies if no predicate is passed.""" 

190 

191 @abstractmethod 

192 def clear_domain(self, domain: str) -> None: 

193 """Clear all cookies for domain and all subdomains.""" 

194 

195 @abstractmethod 

196 def update_cookies(self, cookies: LooseCookies, response_url: URL = URL()) -> None: 

197 """Update cookies.""" 

198 

199 def update_cookies_from_headers( 

200 self, headers: Sequence[str], response_url: URL 

201 ) -> None: 

202 """Update cookies from raw Set-Cookie headers.""" 

203 if headers and (cookies_to_update := parse_set_cookie_headers(headers)): 

204 self.update_cookies(cookies_to_update, response_url) 

205 

206 @abstractmethod 

207 def filter_cookies(self, request_url: URL) -> "BaseCookie[str]": 

208 """Return the jar's cookies filtered by their attributes.""" 

209 

210 

211class AbstractStreamWriter(ABC): 

212 """Abstract stream writer.""" 

213 

214 buffer_size: int = 0 

215 output_size: int = 0 

216 length: int | None = 0 

217 

218 @abstractmethod 

219 async def write(self, chunk: bytes | bytearray | memoryview) -> None: 

220 """Write chunk into stream.""" 

221 

222 @abstractmethod 

223 async def write_eof(self, chunk: bytes = b"") -> None: 

224 """Write last chunk.""" 

225 

226 @abstractmethod 

227 async def drain(self) -> None: 

228 """Flush the write buffer.""" 

229 

230 @abstractmethod 

231 def enable_compression( 

232 self, encoding: str = "deflate", strategy: int | None = None 

233 ) -> None: 

234 """Enable HTTP body compression""" 

235 

236 @abstractmethod 

237 def enable_chunking(self) -> None: 

238 """Enable HTTP chunked mode""" 

239 

240 @abstractmethod 

241 async def write_headers( 

242 self, status_line: str, headers: "CIMultiDict[str]" 

243 ) -> None: 

244 """Write HTTP headers""" 

245 

246 def send_headers(self) -> None: 

247 """Force sending buffered headers if not already sent. 

248 

249 Required only if write_headers() buffers headers instead of sending immediately. 

250 For backwards compatibility, this method does nothing by default. 

251 """ 

252 

253 

254class AbstractAccessLogger(ABC): 

255 """Abstract writer to access log.""" 

256 

257 __slots__ = ("logger", "log_format") 

258 

259 def __init__(self, logger: logging.Logger, log_format: str) -> None: 

260 self.logger = logger 

261 self.log_format = log_format 

262 

263 @abstractmethod 

264 def log(self, request: BaseRequest, response: StreamResponse, time: float) -> None: 

265 """Emit log to logger.""" 

266 

267 @property 

268 def enabled(self) -> bool: 

269 """Check if logger is enabled.""" 

270 return True