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

1import asyncio 

2import logging 

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) 

18 

19from multidict import CIMultiDict 

20from yarl import URL 

21 

22from .helpers import get_running_loop 

23from .typedefs import LooseCookies 

24 

25if TYPE_CHECKING: # pragma: no cover 

26 from .web_app import Application 

27 from .web_exceptions import HTTPException 

28 from .web_request import BaseRequest, Request 

29 from .web_response import StreamResponse 

30else: 

31 BaseRequest = Request = Application = StreamResponse = None 

32 HTTPException = None 

33 

34 

35class AbstractRouter(ABC): 

36 def __init__(self) -> None: 

37 self._frozen = False 

38 

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

40 """Post init stage. 

41 

42 Not an abstract method for sake of backward compatibility, 

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

44 it can override this. 

45 """ 

46 

47 @property 

48 def frozen(self) -> bool: 

49 return self._frozen 

50 

51 def freeze(self) -> None: 

52 """Freeze router.""" 

53 self._frozen = True 

54 

55 @abstractmethod 

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

57 """Return MATCH_INFO for given request""" 

58 

59 

60class AbstractMatchInfo(ABC): 

61 @property # pragma: no branch 

62 @abstractmethod 

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

64 """Execute matched request handler""" 

65 

66 @property 

67 @abstractmethod 

68 def expect_handler( 

69 self, 

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

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

72 

73 @property # pragma: no branch 

74 @abstractmethod 

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

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

77 

78 @abstractmethod # pragma: no branch 

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

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

81 

82 @property # pragma: no branch 

83 @abstractmethod 

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

85 """Stack of nested applications. 

86 

87 Top level application is left-most element. 

88 

89 """ 

90 

91 @abstractmethod 

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

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

94 

95 @abstractmethod 

96 def freeze(self) -> None: 

97 """Freeze the match info. 

98 

99 The method is called after route resolution. 

100 

101 After the call .add_app() is forbidden. 

102 

103 """ 

104 

105 

106class AbstractView(ABC): 

107 """Abstract class based view.""" 

108 

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

110 self._request = request 

111 

112 @property 

113 def request(self) -> Request: 

114 """Request instance.""" 

115 return self._request 

116 

117 @abstractmethod 

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

119 """Execute the view handler.""" 

120 

121 

122class AbstractResolver(ABC): 

123 """Abstract DNS resolver.""" 

124 

125 @abstractmethod 

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

127 """Return IP address for given hostname""" 

128 

129 @abstractmethod 

130 async def close(self) -> None: 

131 """Release resolver""" 

132 

133 

134if TYPE_CHECKING: # pragma: no cover 

135 IterableBase = Iterable[Morsel[str]] 

136else: 

137 IterableBase = Iterable 

138 

139 

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

141 

142 

143class AbstractCookieJar(Sized, IterableBase): 

144 """Abstract Cookie Jar.""" 

145 

146 def __init__(self, *, loop: Optional[asyncio.AbstractEventLoop] = None) -> None: 

147 self._loop = get_running_loop(loop) 

148 

149 @abstractmethod 

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

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

152 

153 @abstractmethod 

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

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

156 

157 @abstractmethod 

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

159 """Update cookies.""" 

160 

161 @abstractmethod 

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

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

164 

165 

166class AbstractStreamWriter(ABC): 

167 """Abstract stream writer.""" 

168 

169 buffer_size = 0 

170 output_size = 0 

171 length: Optional[int] = 0 

172 

173 @abstractmethod 

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

175 """Write chunk into stream.""" 

176 

177 @abstractmethod 

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

179 """Write last chunk.""" 

180 

181 @abstractmethod 

182 async def drain(self) -> None: 

183 """Flush the write buffer.""" 

184 

185 @abstractmethod 

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

187 """Enable HTTP body compression""" 

188 

189 @abstractmethod 

190 def enable_chunking(self) -> None: 

191 """Enable HTTP chunked mode""" 

192 

193 @abstractmethod 

194 async def write_headers( 

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

196 ) -> None: 

197 """Write HTTP headers""" 

198 

199 

200class AbstractAccessLogger(ABC): 

201 """Abstract writer to access log.""" 

202 

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

204 self.logger = logger 

205 self.log_format = log_format 

206 

207 @abstractmethod 

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

209 """Emit log to logger."""