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

1import logging 

2import socket 

3from abc import ABC, abstractmethod 

4from collections.abc import Sized 

5from http.cookies import BaseCookie, Morsel 

6from typing import ( 

7 TYPE_CHECKING, 

8 Any, 

9 Awaitable, 

10 Callable, 

11 Dict, 

12 Generator, 

13 Iterable, 

14 List, 

15 Optional, 

16 Tuple, 

17 TypedDict, 

18 Union, 

19) 

20 

21from multidict import CIMultiDict 

22from yarl import URL 

23 

24from .typedefs import LooseCookies 

25 

26if TYPE_CHECKING: 

27 from .web_app import Application 

28 from .web_exceptions import HTTPException 

29 from .web_request import BaseRequest, Request 

30 from .web_response import StreamResponse 

31else: 

32 BaseRequest = Request = Application = StreamResponse = None 

33 HTTPException = None 

34 

35 

36class AbstractRouter(ABC): 

37 def __init__(self) -> None: 

38 self._frozen = False 

39 

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

41 """Post init stage. 

42 

43 Not an abstract method for sake of backward compatibility, 

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

45 it can override this. 

46 """ 

47 

48 @property 

49 def frozen(self) -> bool: 

50 return self._frozen 

51 

52 def freeze(self) -> None: 

53 """Freeze router.""" 

54 self._frozen = True 

55 

56 @abstractmethod 

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

58 """Return MATCH_INFO for given request""" 

59 

60 

61class AbstractMatchInfo(ABC): 

62 

63 __slots__ = () 

64 

65 @property # pragma: no branch 

66 @abstractmethod 

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

68 """Execute matched request handler""" 

69 

70 @property 

71 @abstractmethod 

72 def expect_handler( 

73 self, 

74 ) -> Callable[[Request], Awaitable[Optional[StreamResponse]]]: 

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

76 

77 @property # pragma: no branch 

78 @abstractmethod 

79 def http_exception(self) -> Optional[HTTPException]: 

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

81 

82 @abstractmethod # pragma: no branch 

83 def get_info(self) -> Dict[str, Any]: # type: ignore[misc] 

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

85 

86 @property # pragma: no branch 

87 @abstractmethod 

88 def apps(self) -> Tuple[Application, ...]: 

89 """Stack of nested applications. 

90 

91 Top level application is left-most element. 

92 

93 """ 

94 

95 @abstractmethod 

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

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

98 

99 @abstractmethod 

100 def freeze(self) -> None: 

101 """Freeze the match info. 

102 

103 The method is called after route resolution. 

104 

105 After the call .add_app() is forbidden. 

106 

107 """ 

108 

109 

110class AbstractView(ABC): 

111 """Abstract class based view.""" 

112 

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

114 self._request = request 

115 

116 @property 

117 def request(self) -> Request: 

118 """Request instance.""" 

119 return self._request 

120 

121 @abstractmethod 

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

123 """Execute the view handler.""" 

124 

125 

126class ResolveResult(TypedDict): 

127 """Resolve result. 

128 

129 This is the result returned from an AbstractResolver's 

130 resolve method. 

131 

132 :param hostname: The hostname that was provided. 

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

134 :param port: The port that was resolved. 

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

136 :param proto: The protocol that was resolved. 

137 :param flags: The flags that were resolved. 

138 """ 

139 

140 hostname: str 

141 host: str 

142 port: int 

143 family: int 

144 proto: int 

145 flags: int 

146 

147 

148class AbstractResolver(ABC): 

149 """Abstract DNS resolver.""" 

150 

151 @abstractmethod 

152 async def resolve( 

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

154 ) -> List[ResolveResult]: 

155 """Return IP address for given hostname""" 

156 

157 @abstractmethod 

158 async def close(self) -> None: 

159 """Release resolver""" 

160 

161 

162if TYPE_CHECKING: 

163 IterableBase = Iterable[Morsel[str]] 

164else: 

165 IterableBase = Iterable 

166 

167 

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

169 

170 

171class AbstractCookieJar(Sized, IterableBase): 

172 """Abstract Cookie Jar.""" 

173 

174 @property 

175 @abstractmethod 

176 def quote_cookie(self) -> bool: 

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

178 

179 @abstractmethod 

180 def clear(self, predicate: Optional[ClearCookiePredicate] = None) -> None: 

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

182 

183 @abstractmethod 

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

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

186 

187 @abstractmethod 

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

189 """Update cookies.""" 

190 

191 @abstractmethod 

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

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

194 

195 

196class AbstractStreamWriter(ABC): 

197 """Abstract stream writer.""" 

198 

199 buffer_size: int = 0 

200 output_size: int = 0 

201 length: Optional[int] = 0 

202 

203 @abstractmethod 

204 async def write( 

205 self, chunk: Union[bytes, bytearray, "memoryview[int]", "memoryview[bytes]"] 

206 ) -> None: 

207 """Write chunk into stream.""" 

208 

209 @abstractmethod 

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

211 """Write last chunk.""" 

212 

213 @abstractmethod 

214 async def drain(self) -> None: 

215 """Flush the write buffer.""" 

216 

217 @abstractmethod 

218 def enable_compression( 

219 self, encoding: str = "deflate", strategy: Optional[int] = None 

220 ) -> None: 

221 """Enable HTTP body compression""" 

222 

223 @abstractmethod 

224 def enable_chunking(self) -> None: 

225 """Enable HTTP chunked mode""" 

226 

227 @abstractmethod 

228 async def write_headers( 

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

230 ) -> None: 

231 """Write HTTP headers""" 

232 

233 

234class AbstractAccessLogger(ABC): 

235 """Abstract writer to access log.""" 

236 

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

238 

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

240 self.logger = logger 

241 self.log_format = log_format 

242 

243 @abstractmethod 

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

245 """Emit log to logger.""" 

246 

247 @property 

248 def enabled(self) -> bool: 

249 """Check if logger is enabled.""" 

250 return True 

251 

252 

253class AbstractAsyncAccessLogger(ABC): 

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

255 

256 __slots__ = () 

257 

258 @abstractmethod 

259 async def log( 

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

261 ) -> None: 

262 """Emit log to logger.""" 

263 

264 @property 

265 def enabled(self) -> bool: 

266 """Check if logger is enabled.""" 

267 return True