1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
|
from __future__ import annotations
from typing import TYPE_CHECKING, Any
from litestar.connection import ASGIConnection
from litestar.enums import ScopeType
from litestar.routes.base import BaseRoute
if TYPE_CHECKING:
from litestar.handlers.asgi_handlers import ASGIRouteHandler
from litestar.types import Receive, Scope, Send
class ASGIRoute(BaseRoute):
"""An ASGI route, handling a single ``ASGIRouteHandler``"""
__slots__ = ("route_handler",)
def __init__(
self,
*,
path: str,
route_handler: ASGIRouteHandler,
) -> None:
"""Initialize the route.
Args:
path: The path for the route.
route_handler: An instance of :class:`~.handlers.ASGIRouteHandler`.
"""
self.route_handler = route_handler
super().__init__(
path=path,
scope_type=ScopeType.ASGI,
handler_names=[route_handler.handler_name],
)
async def handle(self, scope: Scope, receive: Receive, send: Send) -> None:
"""ASGI app that authorizes the connection and then awaits the handler function.
Args:
scope: The ASGI connection scope.
receive: The ASGI receive function.
send: The ASGI send function.
Returns:
None
"""
if self.route_handler.resolve_guards():
connection = ASGIConnection["ASGIRouteHandler", Any, Any, Any](scope=scope, receive=receive)
await self.route_handler.authorize_connection(connection=connection)
await self.route_handler.fn(scope=scope, receive=receive, send=send)
|
