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

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[Any, 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