Coverage for /pythoncovmergedfiles/medio/medio/src/aiohttp/aiohttp/abc.py: 88%

1import logging 

2import socket 

3from abc import ABC, abstractmethod 

4from collections.abc import Awaitable, Callable, Generator, Iterable, Sequence, Sized 

5from http.cookies import BaseCookie, Morsel, SimpleCookie 

6from types import MappingProxyType 

7from typing import TYPE_CHECKING, Any, TypedDict 

8 

9from multidict import CIMultiDict 

10from yarl import URL 

11 

12from ._cookie_helpers import parse_set_cookie_headers 

13from .typedefs import LooseCookies 

14 

15if TYPE_CHECKING: 

16 from .web_app import Application 

17 from .web_exceptions import HTTPException 

18 from .web_request import BaseRequest, Request 

19 from .web_response import StreamResponse 

20else: 

21 BaseRequest = Request = Application = StreamResponse = Any 

22 HTTPException = Any 

23 

24 

25class AbstractRouter(ABC): 

26 def __init__(self) -> None: 

27 self._frozen = False 

28 

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

30 """Post init stage. 

31 

32 Not an abstract method for sake of backward compatibility, 

33 but if the router wants to be aware of the application 

34 it can override this. 

35 """ 

36 

37 @property 

38 def frozen(self) -> bool: 

39 return self._frozen 

40 

41 def freeze(self) -> None: 

42 """Freeze router.""" 

43 self._frozen = True 

44 

45 @abstractmethod 

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

47 """Return MATCH_INFO for given request""" 

48 

49 

50class AbstractMatchInfo(ABC): 

51 

52 __slots__ = () 

53 

54 @property # pragma: no branch 

55 @abstractmethod 

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

57 """Execute matched request handler""" 

58 

59 @property 

60 @abstractmethod 

61 def expect_handler( 

62 self, 

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

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

65 

66 @property # pragma: no branch 

67 @abstractmethod 

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

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

70 

71 @abstractmethod # pragma: no branch 

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

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

74 

75 @property # pragma: no branch 

76 @abstractmethod 

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

78 """Stack of nested applications. 

79 

80 Top level application is left-most element. 

81 

82 """ 

83 

84 @abstractmethod 

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

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

87 

88 @abstractmethod 

89 def freeze(self) -> None: 

90 """Freeze the match info. 

91 

92 The method is called after route resolution. 

93 

94 After the call .add_app() is forbidden. 

95 

96 """ 

97 

98 

99class AbstractView(ABC): 

100 """Abstract class based view.""" 

101 

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

103 self._request = request 

104 

105 @property 

106 def request(self) -> Request: 

107 """Request instance.""" 

108 return self._request 

109 

110 @abstractmethod 

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

112 """Execute the view handler.""" 

113 

114 

115class ResolveResult(TypedDict): 

116 """Resolve result. 

117 

118 This is the result returned from an AbstractResolver's 

119 resolve method. 

120 

121 :param hostname: The hostname that was provided. 

122 :param host: The IP address that was resolved. 

123 :param port: The port that was resolved. 

124 :param family: The address family that was resolved. 

125 :param proto: The protocol that was resolved. 

126 :param flags: The flags that were resolved. 

127 """ 

128 

129 hostname: str 

130 host: str 

131 port: int 

132 family: int 

133 proto: int 

134 flags: int 

135 

136 

137class AbstractResolver(ABC): 

138 """Abstract DNS resolver.""" 

139 

140 @abstractmethod 

141 async def resolve( 

142 self, host: str, port: int = 0, family: socket.AddressFamily = socket.AF_INET 

143 ) -> list[ResolveResult]: 

144 """Return IP address for given hostname""" 

145 

146 @abstractmethod 

147 async def close(self) -> None: 

148 """Release resolver""" 

149 

150 

151ClearCookiePredicate = Callable[[Morsel[str]], bool] 

152 

153 

154class AbstractCookieJar(Sized, Iterable[Morsel[str]]): 

155 """Abstract Cookie Jar.""" 

156 

157 @property 

158 @abstractmethod 

159 def unsafe(self) -> bool: 

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

161 

162 @property 

163 @abstractmethod 

164 def quote_cookie(self) -> bool: 

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

166 

167 @property 

168 @abstractmethod 

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

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

171 

172 @property 

173 @abstractmethod 

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

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

176 

177 @abstractmethod 

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

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

180 

181 @abstractmethod 

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

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

184 

185 @abstractmethod 

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

187 """Update cookies.""" 

188 

189 def update_cookies_from_headers( 

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

191 ) -> None: 

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

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

194 self.update_cookies(cookies_to_update, response_url) 

195 

196 @abstractmethod 

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

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

199 

200 

201class AbstractStreamWriter(ABC): 

202 """Abstract stream writer.""" 

203 

204 buffer_size: int = 0 

205 output_size: int = 0 

206 length: int | None = 0 

207 

208 @abstractmethod 

209 async def write( 

210 self, chunk: "bytes | bytearray | memoryview[int] | memoryview[bytes]" 

211 ) -> None: 

212 """Write chunk into stream.""" 

213 

214 @abstractmethod 

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

216 """Write last chunk.""" 

217 

218 @abstractmethod 

219 async def drain(self) -> None: 

220 """Flush the write buffer.""" 

221 

222 @abstractmethod 

223 def enable_compression( 

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

225 ) -> None: 

226 """Enable HTTP body compression""" 

227 

228 @abstractmethod 

229 def enable_chunking(self) -> None: 

230 """Enable HTTP chunked mode""" 

231 

232 @abstractmethod 

233 async def write_headers(self, status_line: str, headers: CIMultiDict[str]) -> None: 

234 """Write HTTP headers""" 

235 

236 def send_headers(self) -> None: 

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

238 

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

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

241 """ 

242 

243 

244class AbstractAccessLogger(ABC): 

245 """Abstract writer to access log.""" 

246 

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

248 

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

250 self.logger = logger 

251 self.log_format = log_format 

252 

253 @abstractmethod 

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

255 """Emit log to logger.""" 

256 

257 @property 

258 def enabled(self) -> bool: 

259 """Check if logger is enabled.""" 

260 return True 

261 

262 

263class AbstractAsyncAccessLogger(ABC): 

264 """Abstract asynchronous writer to access log.""" 

265 

266 __slots__ = () 

267 

268 @abstractmethod 

269 async def log( 

270 self, request: BaseRequest, response: StreamResponse, request_start: float 

271 ) -> None: 

272 """Emit log to logger.""" 

273 

274 @property 

275 def enabled(self) -> bool: 

276 """Check if logger is enabled.""" 

277 return True