diff options
Diffstat (limited to 'venv/lib/python3.11/site-packages/litestar/_asgi')
16 files changed, 760 insertions, 0 deletions
diff --git a/venv/lib/python3.11/site-packages/litestar/_asgi/__init__.py b/venv/lib/python3.11/site-packages/litestar/_asgi/__init__.py new file mode 100644 index 0000000..4cb42ad --- /dev/null +++ b/venv/lib/python3.11/site-packages/litestar/_asgi/__init__.py @@ -0,0 +1,3 @@ +from litestar._asgi.asgi_router import ASGIRouter + +__all__ = ("ASGIRouter",) diff --git a/venv/lib/python3.11/site-packages/litestar/_asgi/__pycache__/__init__.cpython-311.pyc b/venv/lib/python3.11/site-packages/litestar/_asgi/__pycache__/__init__.cpython-311.pyc Binary files differnew file mode 100644 index 0000000..9b12b52 --- /dev/null +++ b/venv/lib/python3.11/site-packages/litestar/_asgi/__pycache__/__init__.cpython-311.pyc diff --git a/venv/lib/python3.11/site-packages/litestar/_asgi/__pycache__/asgi_router.cpython-311.pyc b/venv/lib/python3.11/site-packages/litestar/_asgi/__pycache__/asgi_router.cpython-311.pyc Binary files differnew file mode 100644 index 0000000..75c2cdb --- /dev/null +++ b/venv/lib/python3.11/site-packages/litestar/_asgi/__pycache__/asgi_router.cpython-311.pyc diff --git a/venv/lib/python3.11/site-packages/litestar/_asgi/__pycache__/utils.cpython-311.pyc b/venv/lib/python3.11/site-packages/litestar/_asgi/__pycache__/utils.cpython-311.pyc Binary files differnew file mode 100644 index 0000000..2e9f267 --- /dev/null +++ b/venv/lib/python3.11/site-packages/litestar/_asgi/__pycache__/utils.cpython-311.pyc diff --git a/venv/lib/python3.11/site-packages/litestar/_asgi/asgi_router.py b/venv/lib/python3.11/site-packages/litestar/_asgi/asgi_router.py new file mode 100644 index 0000000..ebafaf0 --- /dev/null +++ b/venv/lib/python3.11/site-packages/litestar/_asgi/asgi_router.py @@ -0,0 +1,184 @@ +from __future__ import annotations + +import re +from collections import defaultdict +from functools import lru_cache +from traceback import format_exc +from typing import TYPE_CHECKING, Any, Pattern + +from litestar._asgi.routing_trie import validate_node +from litestar._asgi.routing_trie.mapping import add_route_to_trie +from litestar._asgi.routing_trie.traversal import parse_path_to_route +from litestar._asgi.routing_trie.types import create_node +from litestar._asgi.utils import get_route_handlers +from litestar.exceptions import ImproperlyConfiguredException +from litestar.utils import normalize_path + +__all__ = ("ASGIRouter",) + + +if TYPE_CHECKING: + from litestar._asgi.routing_trie.types import RouteTrieNode + from litestar.app import Litestar + from litestar.routes import ASGIRoute, HTTPRoute, WebSocketRoute + from litestar.routes.base import BaseRoute + from litestar.types import ( + ASGIApp, + LifeSpanReceive, + LifeSpanSend, + LifeSpanShutdownCompleteEvent, + LifeSpanShutdownFailedEvent, + LifeSpanStartupCompleteEvent, + LifeSpanStartupFailedEvent, + Method, + Receive, + RouteHandlerType, + Scope, + Send, + ) + + +class ASGIRouter: + """Litestar ASGI router. + + Handling both the ASGI lifespan events and routing of connection requests. + """ + + __slots__ = ( + "_mount_paths_regex", + "_mount_routes", + "_plain_routes", + "_registered_routes", + "_static_routes", + "app", + "root_route_map_node", + "route_handler_index", + "route_mapping", + ) + + def __init__(self, app: Litestar) -> None: + """Initialize ``ASGIRouter``. + + Args: + app: The Litestar app instance + """ + self._mount_paths_regex: Pattern | None = None + self._mount_routes: dict[str, RouteTrieNode] = {} + self._plain_routes: set[str] = set() + self._registered_routes: set[HTTPRoute | WebSocketRoute | ASGIRoute] = set() + self.app = app + self.root_route_map_node: RouteTrieNode = create_node() + self.route_handler_index: dict[str, RouteHandlerType] = {} + self.route_mapping: dict[str, list[BaseRoute]] = defaultdict(list) + + async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None: + """ASGI callable. + + The main entry point to the Router class. + """ + scope.setdefault("path_params", {}) + + path = scope["path"] + if root_path := scope.get("root_path", ""): + path = path.split(root_path, maxsplit=1)[-1] + normalized_path = normalize_path(path) + + asgi_app, scope["route_handler"], scope["path"], scope["path_params"] = self.handle_routing( + path=normalized_path, method=scope.get("method") + ) + await asgi_app(scope, receive, send) + + @lru_cache(1024) # noqa: B019 + def handle_routing(self, path: str, method: Method | None) -> tuple[ASGIApp, RouteHandlerType, str, dict[str, Any]]: + """Handle routing for a given path / method combo. This method is meant to allow easy caching. + + Args: + path: The path of the request. + method: The scope's method, if any. + + Returns: + A tuple composed of the ASGIApp of the route, the route handler instance, the resolved and normalized path and any parsed path params. + """ + return parse_path_to_route( + mount_paths_regex=self._mount_paths_regex, + mount_routes=self._mount_routes, + path=path, + plain_routes=self._plain_routes, + root_node=self.root_route_map_node, + method=method, + ) + + def _store_handler_to_route_mapping(self, route: BaseRoute) -> None: + """Store the mapping of route handlers to routes and to route handler names. + + Args: + route: A Route instance. + + Returns: + None + """ + + for handler in get_route_handlers(route): + if handler.name in self.route_handler_index and str(self.route_handler_index[handler.name]) != str(handler): + raise ImproperlyConfiguredException( + f"route handler names must be unique - {handler.name} is not unique." + ) + identifier = handler.name or str(handler) + self.route_mapping[identifier].append(route) + self.route_handler_index[identifier] = handler + + def construct_routing_trie(self) -> None: + """Create a map of the app's routes. + + This map is used in the asgi router to route requests. + """ + new_routes = [route for route in self.app.routes if route not in self._registered_routes] + for route in new_routes: + add_route_to_trie( + app=self.app, + mount_routes=self._mount_routes, + plain_routes=self._plain_routes, + root_node=self.root_route_map_node, + route=route, + ) + self._store_handler_to_route_mapping(route) + self._registered_routes.add(route) + + validate_node(node=self.root_route_map_node) + if self._mount_routes: + self._mount_paths_regex = re.compile("|".join(sorted(set(self._mount_routes)))) # pyright: ignore + + async def lifespan(self, receive: LifeSpanReceive, send: LifeSpanSend) -> None: + """Handle the ASGI "lifespan" event on application startup and shutdown. + + Args: + receive: The ASGI receive function. + send: The ASGI send function. + + Returns: + None. + """ + + message = await receive() + shutdown_event: LifeSpanShutdownCompleteEvent = {"type": "lifespan.shutdown.complete"} + startup_event: LifeSpanStartupCompleteEvent = {"type": "lifespan.startup.complete"} + + try: + async with self.app.lifespan(): + await send(startup_event) + message = await receive() + + except BaseException as e: + formatted_exception = format_exc() + failure_message: LifeSpanStartupFailedEvent | LifeSpanShutdownFailedEvent + + if message["type"] == "lifespan.startup": + failure_message = {"type": "lifespan.startup.failed", "message": formatted_exception} + else: + failure_message = {"type": "lifespan.shutdown.failed", "message": formatted_exception} + + await send(failure_message) + + raise e + + await send(shutdown_event) diff --git a/venv/lib/python3.11/site-packages/litestar/_asgi/routing_trie/__init__.py b/venv/lib/python3.11/site-packages/litestar/_asgi/routing_trie/__init__.py new file mode 100644 index 0000000..948e394 --- /dev/null +++ b/venv/lib/python3.11/site-packages/litestar/_asgi/routing_trie/__init__.py @@ -0,0 +1,6 @@ +from litestar._asgi.routing_trie.mapping import add_route_to_trie +from litestar._asgi.routing_trie.traversal import parse_path_to_route +from litestar._asgi.routing_trie.types import RouteTrieNode +from litestar._asgi.routing_trie.validate import validate_node + +__all__ = ("RouteTrieNode", "add_route_to_trie", "parse_path_to_route", "validate_node") diff --git a/venv/lib/python3.11/site-packages/litestar/_asgi/routing_trie/__pycache__/__init__.cpython-311.pyc b/venv/lib/python3.11/site-packages/litestar/_asgi/routing_trie/__pycache__/__init__.cpython-311.pyc Binary files differnew file mode 100644 index 0000000..f8658f9 --- /dev/null +++ b/venv/lib/python3.11/site-packages/litestar/_asgi/routing_trie/__pycache__/__init__.cpython-311.pyc diff --git a/venv/lib/python3.11/site-packages/litestar/_asgi/routing_trie/__pycache__/mapping.cpython-311.pyc b/venv/lib/python3.11/site-packages/litestar/_asgi/routing_trie/__pycache__/mapping.cpython-311.pyc Binary files differnew file mode 100644 index 0000000..73323bf --- /dev/null +++ b/venv/lib/python3.11/site-packages/litestar/_asgi/routing_trie/__pycache__/mapping.cpython-311.pyc diff --git a/venv/lib/python3.11/site-packages/litestar/_asgi/routing_trie/__pycache__/traversal.cpython-311.pyc b/venv/lib/python3.11/site-packages/litestar/_asgi/routing_trie/__pycache__/traversal.cpython-311.pyc Binary files differnew file mode 100644 index 0000000..9ed2983 --- /dev/null +++ b/venv/lib/python3.11/site-packages/litestar/_asgi/routing_trie/__pycache__/traversal.cpython-311.pyc diff --git a/venv/lib/python3.11/site-packages/litestar/_asgi/routing_trie/__pycache__/types.cpython-311.pyc b/venv/lib/python3.11/site-packages/litestar/_asgi/routing_trie/__pycache__/types.cpython-311.pyc Binary files differnew file mode 100644 index 0000000..5247c5b --- /dev/null +++ b/venv/lib/python3.11/site-packages/litestar/_asgi/routing_trie/__pycache__/types.cpython-311.pyc diff --git a/venv/lib/python3.11/site-packages/litestar/_asgi/routing_trie/__pycache__/validate.cpython-311.pyc b/venv/lib/python3.11/site-packages/litestar/_asgi/routing_trie/__pycache__/validate.cpython-311.pyc Binary files differnew file mode 100644 index 0000000..57fdcdd --- /dev/null +++ b/venv/lib/python3.11/site-packages/litestar/_asgi/routing_trie/__pycache__/validate.cpython-311.pyc diff --git a/venv/lib/python3.11/site-packages/litestar/_asgi/routing_trie/mapping.py b/venv/lib/python3.11/site-packages/litestar/_asgi/routing_trie/mapping.py new file mode 100644 index 0000000..7a56b97 --- /dev/null +++ b/venv/lib/python3.11/site-packages/litestar/_asgi/routing_trie/mapping.py @@ -0,0 +1,221 @@ +from __future__ import annotations + +from pathlib import Path +from typing import TYPE_CHECKING, Any, cast + +from litestar._asgi.routing_trie.types import ( + ASGIHandlerTuple, + PathParameterSentinel, + create_node, +) +from litestar._asgi.utils import wrap_in_exception_handler +from litestar.types.internal_types import PathParameterDefinition + +__all__ = ("add_mount_route", "add_route_to_trie", "build_route_middleware_stack", "configure_node") + + +if TYPE_CHECKING: + from litestar._asgi.routing_trie.types import RouteTrieNode + from litestar.app import Litestar + from litestar.routes import ASGIRoute, HTTPRoute, WebSocketRoute + from litestar.types import ASGIApp, RouteHandlerType + + +def add_mount_route( + current_node: RouteTrieNode, + mount_routes: dict[str, RouteTrieNode], + root_node: RouteTrieNode, + route: ASGIRoute, +) -> RouteTrieNode: + """Add a node for a mount route. + + Args: + current_node: The current trie node that is being mapped. + mount_routes: A dictionary mapping static routes to trie nodes. + root_node: The root trie node. + route: The route that is being added. + + Returns: + A trie node. + """ + + # we need to ensure that we can traverse the map both through the full path key, e.g. "/my-route/sub-path" and + # via the components keys ["my-route, "sub-path"] + if route.path not in current_node.children: + root_node = current_node + for component in route.path_components: + if component not in current_node.children: + current_node.children[component] = create_node() # type: ignore[index] + current_node = current_node.children[component] # type: ignore[index] + + current_node.is_mount = True + current_node.is_static = route.route_handler.is_static + + if route.path != "/": + mount_routes[route.path] = root_node.children[route.path] = current_node + else: + mount_routes[route.path] = current_node + + return current_node + + +def add_route_to_trie( + app: Litestar, + mount_routes: dict[str, RouteTrieNode], + plain_routes: set[str], + root_node: RouteTrieNode, + route: HTTPRoute | WebSocketRoute | ASGIRoute, +) -> RouteTrieNode: + """Add a new route path (e.g. '/foo/bar/{param:int}') into the route_map tree. + + Inserts non-parameter paths ('plain routes') off the tree's root + node. For paths containing parameters, splits the path on '/' and + nests each path segment under the previous segment's node (see + prefix tree / trie). + + Args: + app: The Litestar app instance. + mount_routes: A dictionary mapping static routes to trie nodes. + plain_routes: A set of routes that do not have path parameters. + root_node: The root trie node. + route: The route that is being added. + + Returns: + A RouteTrieNode instance. + """ + current_node = root_node + + has_path_parameters = bool(route.path_parameters) + + if (route_handler := getattr(route, "route_handler", None)) and getattr(route_handler, "is_mount", False): + current_node = add_mount_route( + current_node=current_node, + mount_routes=mount_routes, + root_node=root_node, + route=cast("ASGIRoute", route), + ) + + elif not has_path_parameters: + plain_routes.add(route.path) + if route.path not in root_node.children: + current_node.children[route.path] = create_node() + current_node = root_node.children[route.path] + + else: + for component in route.path_components: + if isinstance(component, PathParameterDefinition): + current_node.is_path_param_node = True + next_node_key: type[PathParameterSentinel] | str = PathParameterSentinel + + else: + next_node_key = component + + if next_node_key not in current_node.children: + current_node.children[next_node_key] = create_node() + + current_node.child_keys = set(current_node.children.keys()) + current_node = current_node.children[next_node_key] + + if isinstance(component, PathParameterDefinition) and component.type is Path: + current_node.is_path_type = True + + configure_node(route=route, app=app, node=current_node) + return current_node + + +def configure_node( + app: Litestar, + route: HTTPRoute | WebSocketRoute | ASGIRoute, + node: RouteTrieNode, +) -> None: + """Set required attributes and route handlers on route_map tree node. + + Args: + app: The Litestar app instance. + route: The route that is being added. + node: The trie node being configured. + + Returns: + None + """ + from litestar.routes import HTTPRoute, WebSocketRoute + + if not node.path_parameters: + node.path_parameters = {} + + if isinstance(route, HTTPRoute): + for method, handler_mapping in route.route_handler_map.items(): + handler, _ = handler_mapping + node.asgi_handlers[method] = ASGIHandlerTuple( + asgi_app=build_route_middleware_stack(app=app, route=route, route_handler=handler), + handler=handler, + ) + node.path_parameters[method] = route.path_parameters + + elif isinstance(route, WebSocketRoute): + node.asgi_handlers["websocket"] = ASGIHandlerTuple( + asgi_app=build_route_middleware_stack(app=app, route=route, route_handler=route.route_handler), + handler=route.route_handler, + ) + node.path_parameters["websocket"] = route.path_parameters + + else: + node.asgi_handlers["asgi"] = ASGIHandlerTuple( + asgi_app=build_route_middleware_stack(app=app, route=route, route_handler=route.route_handler), + handler=route.route_handler, + ) + node.path_parameters["asgi"] = route.path_parameters + node.is_asgi = True + + +def build_route_middleware_stack( + app: Litestar, + route: HTTPRoute | WebSocketRoute | ASGIRoute, + route_handler: RouteHandlerType, +) -> ASGIApp: + """Construct a middleware stack that serves as the point of entry for each route. + + Args: + app: The Litestar app instance. + route: The route that is being added. + route_handler: The route handler that is being wrapped. + + Returns: + An ASGIApp that is composed of a "stack" of middlewares. + """ + from litestar.middleware.allowed_hosts import AllowedHostsMiddleware + from litestar.middleware.compression import CompressionMiddleware + from litestar.middleware.csrf import CSRFMiddleware + from litestar.middleware.response_cache import ResponseCacheMiddleware + from litestar.routes import HTTPRoute + + # we wrap the route.handle method in the ExceptionHandlerMiddleware + asgi_handler = wrap_in_exception_handler( + app=route.handle, # type: ignore[arg-type] + exception_handlers=route_handler.resolve_exception_handlers(), + ) + + if app.csrf_config: + asgi_handler = CSRFMiddleware(app=asgi_handler, config=app.csrf_config) + + if app.compression_config: + asgi_handler = CompressionMiddleware(app=asgi_handler, config=app.compression_config) + + if isinstance(route, HTTPRoute) and any(r.cache for r in route.route_handlers): + asgi_handler = ResponseCacheMiddleware(app=asgi_handler, config=app.response_cache_config) + + if app.allowed_hosts: + asgi_handler = AllowedHostsMiddleware(app=asgi_handler, config=app.allowed_hosts) + + for middleware in route_handler.resolve_middleware(): + if hasattr(middleware, "__iter__"): + handler, kwargs = cast("tuple[Any, dict[str, Any]]", middleware) + asgi_handler = handler(app=asgi_handler, **kwargs) + else: + asgi_handler = middleware(app=asgi_handler) # type: ignore[call-arg] + + # we wrap the entire stack again in ExceptionHandlerMiddleware + return wrap_in_exception_handler( + app=cast("ASGIApp", asgi_handler), + exception_handlers=route_handler.resolve_exception_handlers(), + ) # pyright: ignore diff --git a/venv/lib/python3.11/site-packages/litestar/_asgi/routing_trie/traversal.py b/venv/lib/python3.11/site-packages/litestar/_asgi/routing_trie/traversal.py new file mode 100644 index 0000000..b7788bd --- /dev/null +++ b/venv/lib/python3.11/site-packages/litestar/_asgi/routing_trie/traversal.py @@ -0,0 +1,170 @@ +from __future__ import annotations + +from functools import lru_cache +from typing import TYPE_CHECKING, Any, Pattern + +from litestar._asgi.routing_trie.types import PathParameterSentinel +from litestar.exceptions import MethodNotAllowedException, NotFoundException +from litestar.utils import normalize_path + +__all__ = ("parse_node_handlers", "parse_path_params", "parse_path_to_route", "traverse_route_map") + + +if TYPE_CHECKING: + from litestar._asgi.routing_trie.types import ASGIHandlerTuple, RouteTrieNode + from litestar.types import ASGIApp, Method, RouteHandlerType + from litestar.types.internal_types import PathParameterDefinition + + +def traverse_route_map( + root_node: RouteTrieNode, + path: str, +) -> tuple[RouteTrieNode, list[str], str]: + """Traverses the application route mapping and retrieves the correct node for the request url. + + Args: + root_node: The root trie node. + path: The request's path. + + Raises: + NotFoundException: If no correlating node is found. + + Returns: + A tuple containing the target RouteMapNode and a list containing all path parameter values. + """ + current_node = root_node + path_params: list[str] = [] + path_components = [p for p in path.split("/") if p] + + for i, component in enumerate(path_components): + if component in current_node.child_keys: + current_node = current_node.children[component] + continue + + if current_node.is_path_param_node: + current_node = current_node.children[PathParameterSentinel] + + if current_node.is_path_type: + path_params.append(normalize_path("/".join(path_components[i:]))) + break + + path_params.append(component) + continue + + raise NotFoundException() + + if not current_node.asgi_handlers: + raise NotFoundException() + + return current_node, path_params, path + + +def parse_node_handlers( + node: RouteTrieNode, + method: Method | None, +) -> ASGIHandlerTuple: + """Retrieve the handler tuple from the node. + + Args: + node: The trie node to parse. + method: The scope's method. + + Raises: + KeyError: If no matching method is found. + + Returns: + An ASGI Handler tuple. + """ + + if node.is_asgi: + return node.asgi_handlers["asgi"] + if method: + return node.asgi_handlers[method] + return node.asgi_handlers["websocket"] + + +@lru_cache(1024) +def parse_path_params( + parameter_definitions: tuple[PathParameterDefinition, ...], path_param_values: tuple[str, ...] +) -> dict[str, Any]: + """Parse path parameters into a dictionary of values. + + Args: + parameter_definitions: The parameter definitions tuple from the route. + path_param_values: The string values extracted from the url + + Raises: + ValueError: If any of path parameters can not be parsed into a value. + + Returns: + A dictionary of parsed path parameters. + """ + return { + param_definition.name: param_definition.parser(value) if param_definition.parser else value + for param_definition, value in zip(parameter_definitions, path_param_values) + } + + +def parse_path_to_route( + method: Method | None, + mount_paths_regex: Pattern | None, + mount_routes: dict[str, RouteTrieNode], + path: str, + plain_routes: set[str], + root_node: RouteTrieNode, +) -> tuple[ASGIApp, RouteHandlerType, str, dict[str, Any]]: + """Given a scope object, retrieve the asgi_handlers and is_mount boolean values from correct trie node. + + Args: + method: The scope's method, if any. + root_node: The root trie node. + path: The path to resolve scope instance. + plain_routes: The set of plain routes. + mount_routes: Mapping of mount routes to trie nodes. + mount_paths_regex: A compiled regex to match the mount routes. + + Raises: + MethodNotAllowedException: if no matching method is found. + NotFoundException: If no correlating node is found or if path params can not be parsed into values according to the node definition. + + Returns: + A tuple containing the stack of middlewares and the route handler that is wrapped by it. + """ + + try: + if path in plain_routes: + asgi_app, handler = parse_node_handlers(node=root_node.children[path], method=method) + return asgi_app, handler, path, {} + + if mount_paths_regex and (match := mount_paths_regex.search(path)): + mount_path = path[match.start() : match.end()] + mount_node = mount_routes[mount_path] + remaining_path = path[match.end() :] + # since we allow regular handlers under static paths, we must validate that the request does not match + # any such handler. + children = [sub_route for sub_route in mount_node.children or [] if sub_route != mount_path] + if not children or all(sub_route not in path for sub_route in children): # type: ignore[operator] + asgi_app, handler = parse_node_handlers(node=mount_node, method=method) + remaining_path = remaining_path or "/" + if not mount_node.is_static: + remaining_path = remaining_path if remaining_path.endswith("/") else f"{remaining_path}/" + return asgi_app, handler, remaining_path, {} + + node, path_parameters, path = traverse_route_map( + root_node=root_node, + path=path, + ) + asgi_app, handler = parse_node_handlers(node=node, method=method) + key = method or ("asgi" if node.is_asgi else "websocket") + parsed_path_parameters = parse_path_params(node.path_parameters[key], tuple(path_parameters)) + + return ( + asgi_app, + handler, + path, + parsed_path_parameters, + ) + except KeyError as e: + raise MethodNotAllowedException() from e + except ValueError as e: + raise NotFoundException() from e diff --git a/venv/lib/python3.11/site-packages/litestar/_asgi/routing_trie/types.py b/venv/lib/python3.11/site-packages/litestar/_asgi/routing_trie/types.py new file mode 100644 index 0000000..d1fc368 --- /dev/null +++ b/venv/lib/python3.11/site-packages/litestar/_asgi/routing_trie/types.py @@ -0,0 +1,85 @@ +from __future__ import annotations + +from dataclasses import dataclass +from typing import TYPE_CHECKING, Literal, NamedTuple + +__all__ = ("ASGIHandlerTuple", "PathParameterSentinel", "RouteTrieNode", "create_node") + + +if TYPE_CHECKING: + from litestar.types import ASGIApp, Method, RouteHandlerType + from litestar.types.internal_types import PathParameterDefinition + + +class PathParameterSentinel: + """Sentinel class designating a path parameter.""" + + +class ASGIHandlerTuple(NamedTuple): + """Encapsulation of a route handler node.""" + + asgi_app: ASGIApp + """An ASGI stack, composed of a handler function and layers of middleware that wrap it.""" + handler: RouteHandlerType + """The route handler instance.""" + + +@dataclass(unsafe_hash=True) +class RouteTrieNode: + """A radix trie node.""" + + __slots__ = ( + "asgi_handlers", + "child_keys", + "children", + "is_asgi", + "is_mount", + "is_static", + "is_path_param_node", + "is_path_type", + "path_parameters", + ) + + asgi_handlers: dict[Method | Literal["websocket", "asgi"], ASGIHandlerTuple] + """A mapping of ASGI handlers stored on the node.""" + child_keys: set[str | type[PathParameterSentinel]] + """ + A set containing the child keys, same as the children dictionary - but as a set, which offers faster lookup. + """ + children: dict[str | type[PathParameterSentinel], RouteTrieNode] + """A dictionary mapping path components or using the PathParameterSentinel class to child nodes.""" + is_path_param_node: bool + """Designates the node as having a path parameter.""" + is_path_type: bool + """Designates the node as having a 'path' type path parameter.""" + is_asgi: bool + """Designate the node as having an `asgi` type handler.""" + is_mount: bool + """Designate the node as being a mount route.""" + is_static: bool + """Designate the node as being a static mount route.""" + path_parameters: dict[Method | Literal["websocket"] | Literal["asgi"], tuple[PathParameterDefinition, ...]] + """A list of tuples containing path parameter definitions. + + This is used for parsing extracted path parameter values. + """ + + +def create_node() -> RouteTrieNode: + """Create a RouteMapNode instance. + + Returns: + A route map node instance. + """ + + return RouteTrieNode( + asgi_handlers={}, + child_keys=set(), + children={}, + is_path_param_node=False, + is_asgi=False, + is_mount=False, + is_static=False, + is_path_type=False, + path_parameters={}, + ) diff --git a/venv/lib/python3.11/site-packages/litestar/_asgi/routing_trie/validate.py b/venv/lib/python3.11/site-packages/litestar/_asgi/routing_trie/validate.py new file mode 100644 index 0000000..5c29fac --- /dev/null +++ b/venv/lib/python3.11/site-packages/litestar/_asgi/routing_trie/validate.py @@ -0,0 +1,47 @@ +from __future__ import annotations + +from itertools import chain +from typing import TYPE_CHECKING + +from litestar.exceptions import ImproperlyConfiguredException + +__all__ = ("validate_node",) + + +if TYPE_CHECKING: + from litestar._asgi.routing_trie.types import RouteTrieNode + + +def validate_node(node: RouteTrieNode) -> None: + """Recursively traverses the trie from the given node upwards. + + Args: + node: A trie node. + + Raises: + ImproperlyConfiguredException + + Returns: + None + """ + if node.is_asgi and bool(set(node.asgi_handlers).difference({"asgi"})): + raise ImproperlyConfiguredException("ASGI handlers must have a unique path not shared by other route handlers.") + + if ( + node.is_mount + and node.children + and any( + chain.from_iterable( + list(child.path_parameters.values()) + if isinstance(child.path_parameters, dict) + else child.path_parameters + for child in node.children.values() + ) + ) + ): + raise ImproperlyConfiguredException("Path parameters are not allowed under a static or mount route.") + + for child in node.children.values(): + if child is node: + continue + validate_node(node=child) diff --git a/venv/lib/python3.11/site-packages/litestar/_asgi/utils.py b/venv/lib/python3.11/site-packages/litestar/_asgi/utils.py new file mode 100644 index 0000000..c4111e0 --- /dev/null +++ b/venv/lib/python3.11/site-packages/litestar/_asgi/utils.py @@ -0,0 +1,44 @@ +from __future__ import annotations + +from typing import TYPE_CHECKING, cast + +__all__ = ("get_route_handlers", "wrap_in_exception_handler") + + +if TYPE_CHECKING: + from litestar.routes import ASGIRoute, HTTPRoute, WebSocketRoute + from litestar.routes.base import BaseRoute + from litestar.types import ASGIApp, ExceptionHandlersMap, RouteHandlerType + + +def wrap_in_exception_handler(app: ASGIApp, exception_handlers: ExceptionHandlersMap) -> ASGIApp: + """Wrap the given ASGIApp in an instance of ExceptionHandlerMiddleware. + + Args: + app: The ASGI app that is being wrapped. + exception_handlers: A mapping of exceptions to handler functions. + + Returns: + A wrapped ASGIApp. + """ + from litestar.middleware.exceptions import ExceptionHandlerMiddleware + + return ExceptionHandlerMiddleware(app=app, exception_handlers=exception_handlers, debug=None) + + +def get_route_handlers(route: BaseRoute) -> list[RouteHandlerType]: + """Retrieve handler(s) as a list for given route. + + Args: + route: The route from which the route handlers are extracted. + + Returns: + The route handlers defined on the route. + """ + route_handlers: list[RouteHandlerType] = [] + if hasattr(route, "route_handlers"): + route_handlers.extend(cast("HTTPRoute", route).route_handlers) + else: + route_handlers.append(cast("WebSocketRoute | ASGIRoute", route).route_handler) + + return route_handlers |