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 

6from typing import TYPE_CHECKING, Any, TypedDict 

7 

8from multidict import CIMultiDict 

9from yarl import URL 

10 

11from ._cookie_helpers import parse_set_cookie_headers 

12from .typedefs import LooseCookies 

13 

14if TYPE_CHECKING: 

15 from .web_app import Application 

16 from .web_exceptions import HTTPException 

17 from .web_request import BaseRequest, Request 

18 from .web_response import StreamResponse 

19else: 

20 BaseRequest = Request = Application = StreamResponse = Any 

21 HTTPException = Any 

22 

23 

24class AbstractRouter(ABC): 

25 def __init__(self) -> None: 

26 self._frozen = False 

27 

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

29 """Post init stage. 

30 

31 Not an abstract method for sake of backward compatibility, 

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

33 it can override this. 

34 """ 

35 

36 @property 

37 def frozen(self) -> bool: 

38 return self._frozen 

39 

40 def freeze(self) -> None: 

41 """Freeze router.""" 

42 self._frozen = True 

43 

44 @abstractmethod 

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

46 """Return MATCH_INFO for given request""" 

47 

48 

49class AbstractMatchInfo(ABC): 

50 

51 __slots__ = () 

52 

53 @property # pragma: no branch 

54 @abstractmethod 

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

56 """Execute matched request handler""" 

57 

58 @property 

59 @abstractmethod 

60 def expect_handler( 

61 self, 

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

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

64 

65 @property # pragma: no branch 

66 @abstractmethod 

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

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

69 

70 @abstractmethod # pragma: no branch 

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

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

73 

74 @property # pragma: no branch 

75 @abstractmethod 

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

77 """Stack of nested applications. 

78 

79 Top level application is left-most element. 

80 

81 """ 

82 

83 @abstractmethod 

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

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

86 

87 @abstractmethod 

88 def freeze(self) -> None: 

89 """Freeze the match info. 

90 

91 The method is called after route resolution. 

92 

93 After the call .add_app() is forbidden. 

94 

95 """ 

96 

97 

98class AbstractView(ABC): 

99 """Abstract class based view.""" 

100 

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

102 self._request = request 

103 

104 @property 

105 def request(self) -> Request: 

106 """Request instance.""" 

107 return self._request 

108 

109 @abstractmethod 

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

111 """Execute the view handler.""" 

112 

113 

114class ResolveResult(TypedDict): 

115 """Resolve result. 

116 

117 This is the result returned from an AbstractResolver's 

118 resolve method. 

119 

120 :param hostname: The hostname that was provided. 

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

122 :param port: The port that was resolved. 

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

124 :param proto: The protocol that was resolved. 

125 :param flags: The flags that were resolved. 

126 """ 

127 

128 hostname: str 

129 host: str 

130 port: int 

131 family: int 

132 proto: int 

133 flags: int 

134 

135 

136class AbstractResolver(ABC): 

137 """Abstract DNS resolver.""" 

138 

139 @abstractmethod 

140 async def resolve( 

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

142 ) -> list[ResolveResult]: 

143 """Return IP address for given hostname""" 

144 

145 @abstractmethod 

146 async def close(self) -> None: 

147 """Release resolver""" 

148 

149 

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

151 

152 

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

154 """Abstract Cookie Jar.""" 

155 

156 @property 

157 @abstractmethod 

158 def unsafe(self) -> bool: 

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

160 

161 @property 

162 @abstractmethod 

163 def quote_cookie(self) -> bool: 

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

165 

166 @abstractmethod 

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

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

169 

170 @abstractmethod 

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

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

173 

174 @abstractmethod 

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

176 """Update cookies.""" 

177 

178 def update_cookies_from_headers( 

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

180 ) -> None: 

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

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

183 self.update_cookies(cookies_to_update, response_url) 

184 

185 @abstractmethod 

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

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

188 

189 

190class AbstractStreamWriter(ABC): 

191 """Abstract stream writer.""" 

192 

193 buffer_size: int = 0 

194 output_size: int = 0 

195 length: int | None = 0 

196 

197 @abstractmethod 

198 async def write( 

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

200 ) -> None: 

201 """Write chunk into stream.""" 

202 

203 @abstractmethod 

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

205 """Write last chunk.""" 

206 

207 @abstractmethod 

208 async def drain(self) -> None: 

209 """Flush the write buffer.""" 

210 

211 @abstractmethod 

212 def enable_compression( 

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

214 ) -> None: 

215 """Enable HTTP body compression""" 

216 

217 @abstractmethod 

218 def enable_chunking(self) -> None: 

219 """Enable HTTP chunked mode""" 

220 

221 @abstractmethod 

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

223 """Write HTTP headers""" 

224 

225 def send_headers(self) -> None: 

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

227 

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

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

230 """ 

231 

232 

233class AbstractAccessLogger(ABC): 

234 """Abstract writer to access log.""" 

235 

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

237 

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

239 self.logger = logger 

240 self.log_format = log_format 

241 

242 @abstractmethod 

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

244 """Emit log to logger.""" 

245 

246 @property 

247 def enabled(self) -> bool: 

248 """Check if logger is enabled.""" 

249 return True 

250 

251 

252class AbstractAsyncAccessLogger(ABC): 

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

254 

255 __slots__ = () 

256 

257 @abstractmethod 

258 async def log( 

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

260 ) -> None: 

261 """Emit log to logger.""" 

262 

263 @property 

264 def enabled(self) -> bool: 

265 """Check if logger is enabled.""" 

266 return True