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(self) -> Callable[[Request], Awaitable[None]]: 

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 def __init__(self, *, loop: Optional[asyncio.AbstractEventLoop] = None) -> None: 

145 self._loop = get_running_loop(loop) 

146 

147 @abstractmethod 

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

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

150 

151 @abstractmethod 

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

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

154 

155 @abstractmethod 

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

157 """Update cookies.""" 

158 

159 @abstractmethod 

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

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

162 

163 

164class AbstractStreamWriter(ABC): 

165 """Abstract stream writer.""" 

166 

167 buffer_size = 0 

168 output_size = 0 

169 length: Optional[int] = 0 

170 

171 @abstractmethod 

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

173 """Write chunk into stream.""" 

174 

175 @abstractmethod 

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

177 """Write last chunk.""" 

178 

179 @abstractmethod 

180 async def drain(self) -> None: 

181 """Flush the write buffer.""" 

182 

183 @abstractmethod 

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

185 """Enable HTTP body compression""" 

186 

187 @abstractmethod 

188 def enable_chunking(self) -> None: 

189 """Enable HTTP chunked mode""" 

190 

191 @abstractmethod 

192 async def write_headers( 

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

194 ) -> None: 

195 """Write HTTP headers""" 

196 

197 

198class AbstractAccessLogger(ABC): 

199 """Abstract writer to access log.""" 

200 

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

202 self.logger = logger 

203 self.log_format = log_format 

204 

205 @abstractmethod 

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

207 """Emit log to logger."""