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

1import logging 

2from abc import ABC, abstractmethod 

3from collections.abc import Sized 

4from http.cookies import BaseCookie, Morsel 

5from typing import ( 

6 TYPE_CHECKING, 

7 Any, 

8 Awaitable, 

9 Callable, 

10 Dict, 

11 Generator, 

12 Iterable, 

13 List, 

14 Optional, 

15 Tuple, 

16) 

17 

18from multidict import CIMultiDict 

19from yarl import URL 

20 

21from .typedefs import LooseCookies 

22 

23if TYPE_CHECKING: # pragma: no cover 

24 from .web_app import Application 

25 from .web_exceptions import HTTPException 

26 from .web_request import BaseRequest, Request 

27 from .web_response import StreamResponse 

28else: 

29 BaseRequest = Request = Application = StreamResponse = None 

30 HTTPException = None 

31 

32 

33class AbstractRouter(ABC): 

34 def __init__(self) -> None: 

35 self._frozen = False 

36 

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

38 """Post init stage. 

39 

40 Not an abstract method for sake of backward compatibility, 

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

42 it can override this. 

43 """ 

44 

45 @property 

46 def frozen(self) -> bool: 

47 return self._frozen 

48 

49 def freeze(self) -> None: 

50 """Freeze router.""" 

51 self._frozen = True 

52 

53 @abstractmethod 

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

55 """Return MATCH_INFO for given request""" 

56 

57 

58class AbstractMatchInfo(ABC): 

59 @property # pragma: no branch 

60 @abstractmethod 

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

62 """Execute matched request handler""" 

63 

64 @property 

65 @abstractmethod 

66 def expect_handler( 

67 self, 

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

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

70 

71 @property # pragma: no branch 

72 @abstractmethod 

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

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

75 

76 @abstractmethod # pragma: no branch 

77 def get_info(self) -> Dict[str, Any]: 

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

79 

80 @property # pragma: no branch 

81 @abstractmethod 

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

83 """Stack of nested applications. 

84 

85 Top level application is left-most element. 

86 

87 """ 

88 

89 @abstractmethod 

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

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

92 

93 @abstractmethod 

94 def freeze(self) -> None: 

95 """Freeze the match info. 

96 

97 The method is called after route resolution. 

98 

99 After the call .add_app() is forbidden. 

100 

101 """ 

102 

103 

104class AbstractView(ABC): 

105 """Abstract class based view.""" 

106 

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

108 self._request = request 

109 

110 @property 

111 def request(self) -> Request: 

112 """Request instance.""" 

113 return self._request 

114 

115 @abstractmethod 

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

117 """Execute the view handler.""" 

118 

119 

120class AbstractResolver(ABC): 

121 """Abstract DNS resolver.""" 

122 

123 @abstractmethod 

124 async def resolve(self, host: str, port: int, family: int) -> List[Dict[str, Any]]: 

125 """Return IP address for given hostname""" 

126 

127 @abstractmethod 

128 async def close(self) -> None: 

129 """Release resolver""" 

130 

131 

132if TYPE_CHECKING: # pragma: no cover 

133 IterableBase = Iterable[Morsel[str]] 

134else: 

135 IterableBase = Iterable 

136 

137 

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

139 

140 

141class AbstractCookieJar(Sized, IterableBase): 

142 """Abstract Cookie Jar.""" 

143 

144 @abstractmethod 

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

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

147 

148 @abstractmethod 

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

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

151 

152 @abstractmethod 

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

154 """Update cookies.""" 

155 

156 @abstractmethod 

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

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

159 

160 

161class AbstractStreamWriter(ABC): 

162 """Abstract stream writer.""" 

163 

164 buffer_size = 0 

165 output_size = 0 

166 length: Optional[int] = 0 

167 

168 @abstractmethod 

169 async def write(self, chunk: bytes) -> None: 

170 """Write chunk into stream.""" 

171 

172 @abstractmethod 

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

174 """Write last chunk.""" 

175 

176 @abstractmethod 

177 async def drain(self) -> None: 

178 """Flush the write buffer.""" 

179 

180 @abstractmethod 

181 def enable_compression(self, encoding: str = "deflate") -> None: 

182 """Enable HTTP body compression""" 

183 

184 @abstractmethod 

185 def enable_chunking(self) -> None: 

186 """Enable HTTP chunked mode""" 

187 

188 @abstractmethod 

189 async def write_headers( 

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

191 ) -> None: 

192 """Write HTTP headers""" 

193 

194 

195class AbstractAccessLogger(ABC): 

196 """Abstract writer to access log.""" 

197 

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

199 self.logger = logger 

200 self.log_format = log_format 

201 

202 @abstractmethod 

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

204 """Emit log to logger.""" 

205 

206 

207class AbstractAsyncAccessLogger(ABC): 

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

209 

210 @abstractmethod 

211 async def log( 

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

213 ) -> None: 

214 """Emit log to logger."""