codekingpro/portable-devtools
115k
1import asyncio
2import logging
3import warnings
4from functools import lru_cache, partial, update_wrapper
5from typing import (
6 TYPE_CHECKING,
7 Any,
8 AsyncIterator,
9 Awaitable,
10 Callable,
11 Dict,
12 Iterable,
13 Iterator,
14 List,
15 Mapping,
16 MutableMapping,
17 Optional,
18 Sequence,
19 Tuple,
20 Type,
21 TypeVar,
22 Union,
23 cast,
24 overload,
25)
26
27from aiosignal import Signal
28from frozenlist import FrozenList
29
30from . import hdrs
31from .abc import (
32 AbstractAccessLogger,
33 AbstractMatchInfo,
34 AbstractRouter,
35 AbstractStreamWriter,
36)
37from .helpers import DEBUG, AppKey
38from .http_parser import RawRequestMessage
39from .log import web_logger
40from .streams import StreamReader
41from .typedefs import Handler, Middleware
42from .web_exceptions import NotAppKeyWarning
43from .web_log import AccessLogger
44from .web_middlewares import _fix_request_current_app
45from .web_protocol import RequestHandler
46from .web_request import Request
47from .web_response import StreamResponse
48from .web_routedef import AbstractRouteDef
49from .web_server import Server
50from .web_urldispatcher import (
51 AbstractResource,
52 AbstractRoute,
53 Domain,
54 MaskDomain,
55 MatchedSubAppResource,
56 PrefixedSubAppResource,
57 SystemRoute,
58 UrlDispatcher,
59)
60
61__all__ = ("Application", "CleanupError")
62
63
64if TYPE_CHECKING:
65 _AppSignal = Signal["Application"]
66 _RespPrepareSignal = Signal[Request, StreamResponse]
67 _Middlewares = FrozenList[Middleware]
68 _MiddlewaresHandlers = Optional[Sequence[Tuple[Middleware, bool]]]
69 _Subapps = List["Application"]
70else:
71 # No type checker mode, skip types
72 _AppSignal = Signal
73 _RespPrepareSignal = Signal
74 _Middlewares = FrozenList
75 _MiddlewaresHandlers = Optional[Sequence]
76 _Subapps = List
77
78_T = TypeVar("_T")
79_U = TypeVar("_U")
80_Resource = TypeVar("_Resource", bound=AbstractResource)
81
82
83def _build_middlewares(
84 handler: Handler, apps: Tuple["Application", ...]
85) -> Callable[[Request], Awaitable[StreamResponse]]:
86 """Apply middlewares to handler."""
87 for app in apps[::-1]:
88 for m, _ in app._middlewares_handlers: # type: ignore[union-attr]
89 handler = update_wrapper(partial(m, handler=handler), handler)
90 return handler
91
92
93_cached_build_middleware = lru_cache(maxsize=1024)(_build_middlewares)
94
95
96class Application(MutableMapping[Union[str, AppKey[Any]], Any]):
97 ATTRS = frozenset(
98 [
99 "logger",
100 "_debug",
101 "_router",
102 "_loop",
103 "_handler_args",
104 "_middlewares",
105 "_middlewares_handlers",
106 "_has_legacy_middlewares",
107 "_run_middlewares",
108 "_state",
109 "_frozen",
110 "_pre_frozen",
111 "_subapps",
112 "_on_response_prepare",
113 "_on_startup",
114 "_on_shutdown",
115 "_on_cleanup",
116 "_client_max_size",
117 "_cleanup_ctx",
118 ]
119 )
120
121 def __init__(
122 self,
123 *,
124 logger: logging.Logger = web_logger,
125 router: Optional[UrlDispatcher] = None,
126 middlewares: Iterable[Middleware] = (),
127 handler_args: Optional[Mapping[str, Any]] = None,
128 client_max_size: int = 1024**2,
129 loop: Optional[asyncio.AbstractEventLoop] = None,
130 debug: Any = ..., # mypy doesn't support ellipsis
131 ) -> None:
132 if router is None:
133 router = UrlDispatcher()
134 else:
135 warnings.warn(
136 "router argument is deprecated", DeprecationWarning, stacklevel=2
137 )
138 assert isinstance(router, AbstractRouter), router
139
140 if loop is not None:
141 warnings.warn(
142 "loop argument is deprecated", DeprecationWarning, stacklevel=2
143 )
144
145 if debug is not ...:
146 warnings.warn(
147 "debug argument is deprecated", DeprecationWarning, stacklevel=2
148 )
149 self._debug = debug
150 self._router: UrlDispatcher = router
151 self._loop = loop
152 self._handler_args = handler_args
153 self.logger = logger
154
155 self._middlewares: _Middlewares = FrozenList(middlewares)
156
157 # initialized on freezing
158 self._middlewares_handlers: _MiddlewaresHandlers = None
159 # initialized on freezing
160 self._run_middlewares: Optional[bool] = None
161 self._has_legacy_middlewares: bool = True
162
163 self._state: Dict[Union[AppKey[Any], str], object] = {}
164 self._frozen = False
165 self._pre_frozen = False
166 self._subapps: _Subapps = []
167
168 self._on_response_prepare: _RespPrepareSignal = Signal(self)
169 self._on_startup: _AppSignal = Signal(self)
170 self._on_shutdown: _AppSignal = Signal(self)
171 self._on_cleanup: _AppSignal = Signal(self)
172 self._cleanup_ctx = CleanupContext()
173 self._on_startup.append(self._cleanup_ctx._on_startup)
174 self._on_cleanup.append(self._cleanup_ctx._on_cleanup)
175 self._client_max_size = client_max_size
176
177 def __init_subclass__(cls: Type["Application"]) -> None:
178 warnings.warn(
179 "Inheritance class {} from web.Application "
180 "is discouraged".format(cls.__name__),
181 DeprecationWarning,
182 stacklevel=3,
183 )
184
185 if DEBUG: # pragma: no cover
186
187 def __setattr__(self, name: str, val: Any) -> None:
188 if name not in self.ATTRS:
189 warnings.warn(
190 "Setting custom web.Application.{} attribute "
191 "is discouraged".format(name),
192 DeprecationWarning,
193 stacklevel=2,
194 )
195 super().__setattr__(name, val)
196
197 # MutableMapping API
198
199 def __eq__(self, other: object) -> bool:
200 return self is other
201
202 @overload # type: ignore[override]
203 def __getitem__(self, key: AppKey[_T]) -> _T: ...
204
205 @overload
206 def __getitem__(self, key: str) -> Any: ...
207
208 def __getitem__(self, key: Union[str, AppKey[_T]]) -> Any:
209 return self._state[key]
210
211 def _check_frozen(self) -> None:
212 if self._frozen:
213 warnings.warn(
214 "Changing state of started or joined application is deprecated",
215 DeprecationWarning,
216 stacklevel=3,
217 )
218
219 @overload # type: ignore[override]
220 def __setitem__(self, key: AppKey[_T], value: _T) -> None: ...
221
222 @overload
223 def __setitem__(self, key: str, value: Any) -> None: ...
224
225 def __setitem__(self, key: Union[str, AppKey[_T]], value: Any) -> None:
226 self._check_frozen()
227 if not isinstance(key, AppKey):
228 warnings.warn(
229 "It is recommended to use web.AppKey instances for keys.\n"
230 + "https://docs.aiohttp.org/en/stable/web_advanced.html"
231 + "#application-s-config",
232 category=NotAppKeyWarning,
233 stacklevel=2,
234 )
235 self._state[key] = value
236
237 def __delitem__(self, key: Union[str, AppKey[_T]]) -> None:
238 self._check_frozen()
239 del self._state[key]
240
241 def __len__(self) -> int:
242 return len(self._state)
243
244 def __iter__(self) -> Iterator[Union[str, AppKey[Any]]]:
245 return iter(self._state)
246
247 def __hash__(self) -> int:
248 return id(self)
249
250 @overload # type: ignore[override]
251 def get(self, key: AppKey[_T], default: None = ...) -> Optional[_T]: ...
252
253 @overload
254 def get(self, key: AppKey[_T], default: _U) -> Union[_T, _U]: ...
255
256 @overload
257 def get(self, key: str, default: Any = ...) -> Any: ...
258
259 def get(self, key: Union[str, AppKey[_T]], default: Any = None) -> Any:
260 return self._state.get(key, default)
261
262 ########
263 @property
264 def loop(self) -> asyncio.AbstractEventLoop:
265 # Technically the loop can be None
266 # but we mask it by explicit type cast
267 # to provide more convenient type annotation
268 warnings.warn("loop property is deprecated", DeprecationWarning, stacklevel=2)
269 return cast(asyncio.AbstractEventLoop, self._loop)
270
271 def _set_loop(self, loop: Optional[asyncio.AbstractEventLoop]) -> None:
272 if loop is None:
273 loop = asyncio.get_event_loop()
274 if self._loop is not None and self._loop is not loop:
275 raise RuntimeError(
276 "web.Application instance initialized with different loop"
277 )
278
279 self._loop = loop
280
281 # set loop debug
282 if self._debug is ...:
283 self._debug = loop.get_debug()
284
285 # set loop to sub applications
286 for subapp in self._subapps:
287 subapp._set_loop(loop)
288
289 @property
290 def pre_frozen(self) -> bool:
291 return self._pre_frozen
292
293 def pre_freeze(self) -> None:
294 if self._pre_frozen:
295 return
296
297 self._pre_frozen = True
298 self._middlewares.freeze()
299 self._router.freeze()
300 self._on_response_prepare.freeze()
301 self._cleanup_ctx.freeze()
302 self._on_startup.freeze()
303 self._on_shutdown.freeze()
304 self._on_cleanup.freeze()
305 self._middlewares_handlers = tuple(self._prepare_middleware())
306 self._has_legacy_middlewares = any(
307 not new_style for _, new_style in self._middlewares_handlers
308 )
309
310 # If current app and any subapp do not have middlewares avoid run all
311 # of the code footprint that it implies, which have a middleware
312 # hardcoded per app that sets up the current_app attribute. If no
313 # middlewares are configured the handler will receive the proper
314 # current_app without needing all of this code.
315 self._run_middlewares = True if self.middlewares else False
316
317 for subapp in self._subapps:
318 subapp.pre_freeze()
319 self._run_middlewares = self._run_middlewares or subapp._run_middlewares
320
321 @property
322 def frozen(self) -> bool:
323 return self._frozen
324
325 def freeze(self) -> None:
326 if self._frozen:
327 return
328
329 self.pre_freeze()
330 self._frozen = True
331 for subapp in self._subapps:
332 subapp.freeze()
333
334 @property
335 def debug(self) -> bool:
336 warnings.warn("debug property is deprecated", DeprecationWarning, stacklevel=2)
337 return self._debug # type: ignore[no-any-return]
338
339 def _reg_subapp_signals(self, subapp: "Application") -> None:
340 def reg_handler(signame: str) -> None:
341 subsig = getattr(subapp, signame)
342
343 async def handler(app: "Application") -> None:
344 await subsig.send(subapp)
345
346 appsig = getattr(self, signame)
347 appsig.append(handler)
348
349 reg_handler("on_startup")
350 reg_handler("on_shutdown")
351 reg_handler("on_cleanup")
352
353 def add_subapp(self, prefix: str, subapp: "Application") -> PrefixedSubAppResource:
354 if not isinstance(prefix, str):
355 raise TypeError("Prefix must be str")
356 prefix = prefix.rstrip("/")
357 if not prefix:
358 raise ValueError("Prefix cannot be empty")
359 factory = partial(PrefixedSubAppResource, prefix, subapp)
360 return self._add_subapp(factory, subapp)
361
362 def _add_subapp(
363 self, resource_factory: Callable[[], _Resource], subapp: "Application"
364 ) -> _Resource:
365 if self.frozen:
366 raise RuntimeError("Cannot add sub application to frozen application")
367 if subapp.frozen:
368 raise RuntimeError("Cannot add frozen application")
369 resource = resource_factory()
370 self.router.register_resource(resource)
371 self._reg_subapp_signals(subapp)
372 self._subapps.append(subapp)
373 subapp.pre_freeze()
374 if self._loop is not None:
375 subapp._set_loop(self._loop)
376 return resource
377
378 def add_domain(self, domain: str, subapp: "Application") -> MatchedSubAppResource:
379 if not isinstance(domain, str):
380 raise TypeError("Domain must be str")
381 elif "*" in domain:
382 rule: Domain = MaskDomain(domain)
383 else:
384 rule = Domain(domain)
385 factory = partial(MatchedSubAppResource, rule, subapp)
386 return self._add_subapp(factory, subapp)
387
388 def add_routes(self, routes: Iterable[AbstractRouteDef]) -> List[AbstractRoute]:
389 return self.router.add_routes(routes)
390
391 @property
392 def on_response_prepare(self) -> _RespPrepareSignal:
393 return self._on_response_prepare
394
395 @property
396 def on_startup(self) -> _AppSignal:
397 return self._on_startup
398
399 @property
400 def on_shutdown(self) -> _AppSignal:
401 return self._on_shutdown
402
403 @property
404 def on_cleanup(self) -> _AppSignal:
405 return self._on_cleanup
406
407 @property
408 def cleanup_ctx(self) -> "CleanupContext":
409 return self._cleanup_ctx
410
411 @property
412 def router(self) -> UrlDispatcher:
413 return self._router
414
415 @property
416 def middlewares(self) -> _Middlewares:
417 return self._middlewares
418
419 def _make_handler(
420 self,
421 *,
422 loop: Optional[asyncio.AbstractEventLoop] = None,
423 access_log_class: Type[AbstractAccessLogger] = AccessLogger,
424 **kwargs: Any,
425 ) -> Server:
426
427 if not issubclass(access_log_class, AbstractAccessLogger):
428 raise TypeError(
429 "access_log_class must be subclass of "
430 "aiohttp.abc.AbstractAccessLogger, got {}".format(access_log_class)
431 )
432
433 self._set_loop(loop)
434 self.freeze()
435
436 kwargs["debug"] = self._debug
437 kwargs["access_log_class"] = access_log_class
438 if self._handler_args:
439 for k, v in self._handler_args.items():
440 kwargs[k] = v
441
442 return Server(
443 self._handle, # type: ignore[arg-type]
444 request_factory=self._make_request,
445 loop=self._loop,
446 **kwargs,
447 )
448
449 def make_handler(
450 self,
451 *,
452 loop: Optional[asyncio.AbstractEventLoop] = None,
453 access_log_class: Type[AbstractAccessLogger] = AccessLogger,
454 **kwargs: Any,
455 ) -> Server:
456
457 warnings.warn(
458 "Application.make_handler(...) is deprecated, use AppRunner API instead",
459 DeprecationWarning,
460 stacklevel=2,
461 )
462
463 return self._make_handler(
464 loop=loop, access_log_class=access_log_class, **kwargs
465 )
466
467 async def startup(self) -> None:
468 """Causes on_startup signal
469
470 Should be called in the event loop along with the request handler.
471 """
472 await self.on_startup.send(self)
473
474 async def shutdown(self) -> None:
475 """Causes on_shutdown signal
476
477 Should be called before cleanup()
478 """
479 await self.on_shutdown.send(self)
480
481 async def cleanup(self) -> None:
482 """Causes on_cleanup signal
483
484 Should be called after shutdown()
485 """
486 if self.on_cleanup.frozen:
487 await self.on_cleanup.send(self)
488 else:
489 # If an exception occurs in startup, ensure cleanup contexts are completed.
490 await self._cleanup_ctx._on_cleanup(self)
491
492 def _make_request(
493 self,
494 message: RawRequestMessage,
495 payload: StreamReader,
496 protocol: RequestHandler,
497 writer: AbstractStreamWriter,
498 task: "asyncio.Task[None]",
499 _cls: Type[Request] = Request,
500 ) -> Request:
501 if TYPE_CHECKING:
502 assert self._loop is not None
503 return _cls(
504 message,
505 payload,
506 protocol,
507 writer,
508 task,
509 self._loop,
510 client_max_size=self._client_max_size,
511 )
512
513 def _prepare_middleware(self) -> Iterator[Tuple[Middleware, bool]]:
514 for m in reversed(self._middlewares):
515 if getattr(m, "__middleware_version__", None) == 1:
516 yield m, True
517 else:
518 warnings.warn(
519 f'old-style middleware "{m!r}" deprecated, see #2252',
520 DeprecationWarning,
521 stacklevel=2,
522 )
523 yield m, False
524
525 yield _fix_request_current_app(self), True
526
527 async def _handle(self, request: Request) -> StreamResponse:
528 loop = asyncio.get_event_loop()
529 debug = loop.get_debug()
530 match_info = await self._router.resolve(request)
531 if debug: # pragma: no cover
532 if not isinstance(match_info, AbstractMatchInfo):
533 raise TypeError(
534 "match_info should be AbstractMatchInfo "
535 "instance, not {!r}".format(match_info)
536 )
537 match_info.add_app(self)
538
539 match_info.freeze()
540
541 request._match_info = match_info
542
543 if request.headers.get(hdrs.EXPECT):
544 resp = await match_info.expect_handler(request)
545 await request.writer.drain()
546 if resp is not None:
547 return resp
548
549 handler = match_info.handler
550
551 if self._run_middlewares:
552 # If its a SystemRoute, don't cache building the middlewares since
553 # they are constructed for every MatchInfoError as a new handler
554 # is made each time.
555 if not self._has_legacy_middlewares and not isinstance(
556 match_info.route, SystemRoute
557 ):
558 handler = _cached_build_middleware(handler, match_info.apps)
559 else:
560 for app in match_info.apps[::-1]:
561 for m, new_style in app._middlewares_handlers: # type: ignore[union-attr]
562 if new_style:
563 handler = update_wrapper(
564 partial(m, handler=handler), handler
565 )
566 else:
567 handler = await m(app, handler) # type: ignore[arg-type,assignment]
568
569 return await handler(request)
570
571 def __call__(self) -> "Application":
572 """gunicorn compatibility"""
573 return self
574
575 def __repr__(self) -> str:
576 return f"<Application 0x{id(self):x}>"
577
578 def __bool__(self) -> bool:
579 return True
580
581
582class CleanupError(RuntimeError):
583 @property
584 def exceptions(self) -> List[BaseException]:
585 return cast(List[BaseException], self.args[1])
586
587
588if TYPE_CHECKING:
589 _CleanupContextBase = FrozenList[Callable[[Application], AsyncIterator[None]]]
590else:
591 _CleanupContextBase = FrozenList
592
593
594class CleanupContext(_CleanupContextBase):
595 def __init__(self) -> None:
596 super().__init__()
597 self._exits: List[AsyncIterator[None]] = []
598
599 async def _on_startup(self, app: Application) -> None:
600 for cb in self:
601 it = cb(app).__aiter__()
602 await it.__anext__()
603 self._exits.append(it)
604
605 async def _on_cleanup(self, app: Application) -> None:
606 errors = []
607 for it in reversed(self._exits):
608 try:
609 await it.__anext__()
610 except StopAsyncIteration:
611 pass
612 except (Exception, asyncio.CancelledError) as exc:
613 errors.append(exc)
614 else:
615 errors.append(RuntimeError(f"{it!r} has more than one 'yield'"))
616 if errors:
617 if len(errors) == 1:
618 raise errors[0]
619 else:
620 raise CleanupError("Multiple errors on cleanup stage", errors)
621