diff options
author | cyfraeviolae <cyfraeviolae> | 2024-04-03 03:17:55 -0400 |
---|---|---|
committer | cyfraeviolae <cyfraeviolae> | 2024-04-03 03:17:55 -0400 |
commit | 12cf076118570eebbff08c6b3090e0d4798447a1 (patch) | |
tree | 3ba25e17e3c3a5e82316558ba3864b955919ff72 /venv/lib/python3.11/site-packages/litestar/_asgi | |
parent | c45662ff3923b34614ddcc8feb9195541166dcc5 (diff) |
no venv
Diffstat (limited to 'venv/lib/python3.11/site-packages/litestar/_asgi')
16 files changed, 0 insertions, 760 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 deleted file mode 100644 index 4cb42ad..0000000 --- a/venv/lib/python3.11/site-packages/litestar/_asgi/__init__.py +++ /dev/null @@ -1,3 +0,0 @@ -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 differdeleted file mode 100644 index 9b12b52..0000000 --- a/venv/lib/python3.11/site-packages/litestar/_asgi/__pycache__/__init__.cpython-311.pyc +++ /dev/null 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 differdeleted file mode 100644 index 75c2cdb..0000000 --- a/venv/lib/python3.11/site-packages/litestar/_asgi/__pycache__/asgi_router.cpython-311.pyc +++ /dev/null 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 differdeleted file mode 100644 index 2e9f267..0000000 --- a/venv/lib/python3.11/site-packages/litestar/_asgi/__pycache__/utils.cpython-311.pyc +++ /dev/null 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 deleted file mode 100644 index ebafaf0..0000000 --- a/venv/lib/python3.11/site-packages/litestar/_asgi/asgi_router.py +++ /dev/null @@ -1,184 +0,0 @@ -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 deleted file mode 100644 index 948e394..0000000 --- a/venv/lib/python3.11/site-packages/litestar/_asgi/routing_trie/__init__.py +++ /dev/null @@ -1,6 +0,0 @@ -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 differdeleted file mode 100644 index f8658f9..0000000 --- a/venv/lib/python3.11/site-packages/litestar/_asgi/routing_trie/__pycache__/__init__.cpython-311.pyc +++ /dev/null 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 differdeleted file mode 100644 index 73323bf..0000000 --- a/venv/lib/python3.11/site-packages/litestar/_asgi/routing_trie/__pycache__/mapping.cpython-311.pyc +++ /dev/null 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 differdeleted file mode 100644 index 9ed2983..0000000 --- a/venv/lib/python3.11/site-packages/litestar/_asgi/routing_trie/__pycache__/traversal.cpython-311.pyc +++ /dev/null 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 differdeleted file mode 100644 index 5247c5b..0000000 --- a/venv/lib/python3.11/site-packages/litestar/_asgi/routing_trie/__pycache__/types.cpython-311.pyc +++ /dev/null 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 differdeleted file mode 100644 index 57fdcdd..0000000 --- a/venv/lib/python3.11/site-packages/litestar/_asgi/routing_trie/__pycache__/validate.cpython-311.pyc +++ /dev/null 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 deleted file mode 100644 index 7a56b97..0000000 --- a/venv/lib/python3.11/site-packages/litestar/_asgi/routing_trie/mapping.py +++ /dev/null @@ -1,221 +0,0 @@ -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 deleted file mode 100644 index b7788bd..0000000 --- a/venv/lib/python3.11/site-packages/litestar/_asgi/routing_trie/traversal.py +++ /dev/null @@ -1,170 +0,0 @@ -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 deleted file mode 100644 index d1fc368..0000000 --- a/venv/lib/python3.11/site-packages/litestar/_asgi/routing_trie/types.py +++ /dev/null @@ -1,85 +0,0 @@ -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 deleted file mode 100644 index 5c29fac..0000000 --- a/venv/lib/python3.11/site-packages/litestar/_asgi/routing_trie/validate.py +++ /dev/null @@ -1,47 +0,0 @@ -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 deleted file mode 100644 index c4111e0..0000000 --- a/venv/lib/python3.11/site-packages/litestar/_asgi/utils.py +++ /dev/null @@ -1,44 +0,0 @@ -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 |