summaryrefslogtreecommitdiff
path: root/venv/lib/python3.11/site-packages/litestar/_asgi
diff options
context:
space:
mode:
authorcyfraeviolae <cyfraeviolae>2024-04-03 03:17:55 -0400
committercyfraeviolae <cyfraeviolae>2024-04-03 03:17:55 -0400
commit12cf076118570eebbff08c6b3090e0d4798447a1 (patch)
tree3ba25e17e3c3a5e82316558ba3864b955919ff72 /venv/lib/python3.11/site-packages/litestar/_asgi
parentc45662ff3923b34614ddcc8feb9195541166dcc5 (diff)
no venv
Diffstat (limited to 'venv/lib/python3.11/site-packages/litestar/_asgi')
-rw-r--r--venv/lib/python3.11/site-packages/litestar/_asgi/__init__.py3
-rw-r--r--venv/lib/python3.11/site-packages/litestar/_asgi/__pycache__/__init__.cpython-311.pycbin289 -> 0 bytes
-rw-r--r--venv/lib/python3.11/site-packages/litestar/_asgi/__pycache__/asgi_router.cpython-311.pycbin9201 -> 0 bytes
-rw-r--r--venv/lib/python3.11/site-packages/litestar/_asgi/__pycache__/utils.cpython-311.pycbin2149 -> 0 bytes
-rw-r--r--venv/lib/python3.11/site-packages/litestar/_asgi/asgi_router.py184
-rw-r--r--venv/lib/python3.11/site-packages/litestar/_asgi/routing_trie/__init__.py6
-rw-r--r--venv/lib/python3.11/site-packages/litestar/_asgi/routing_trie/__pycache__/__init__.cpython-311.pycbin614 -> 0 bytes
-rw-r--r--venv/lib/python3.11/site-packages/litestar/_asgi/routing_trie/__pycache__/mapping.cpython-311.pycbin8784 -> 0 bytes
-rw-r--r--venv/lib/python3.11/site-packages/litestar/_asgi/routing_trie/__pycache__/traversal.cpython-311.pycbin7826 -> 0 bytes
-rw-r--r--venv/lib/python3.11/site-packages/litestar/_asgi/routing_trie/__pycache__/types.cpython-311.pycbin2798 -> 0 bytes
-rw-r--r--venv/lib/python3.11/site-packages/litestar/_asgi/routing_trie/__pycache__/validate.cpython-311.pycbin2362 -> 0 bytes
-rw-r--r--venv/lib/python3.11/site-packages/litestar/_asgi/routing_trie/mapping.py221
-rw-r--r--venv/lib/python3.11/site-packages/litestar/_asgi/routing_trie/traversal.py170
-rw-r--r--venv/lib/python3.11/site-packages/litestar/_asgi/routing_trie/types.py85
-rw-r--r--venv/lib/python3.11/site-packages/litestar/_asgi/routing_trie/validate.py47
-rw-r--r--venv/lib/python3.11/site-packages/litestar/_asgi/utils.py44
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
deleted file mode 100644
index 9b12b52..0000000
--- a/venv/lib/python3.11/site-packages/litestar/_asgi/__pycache__/__init__.cpython-311.pyc
+++ /dev/null
Binary files differ
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
deleted file mode 100644
index 75c2cdb..0000000
--- a/venv/lib/python3.11/site-packages/litestar/_asgi/__pycache__/asgi_router.cpython-311.pyc
+++ /dev/null
Binary files differ
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
deleted file mode 100644
index 2e9f267..0000000
--- a/venv/lib/python3.11/site-packages/litestar/_asgi/__pycache__/utils.cpython-311.pyc
+++ /dev/null
Binary files differ
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
deleted file mode 100644
index f8658f9..0000000
--- a/venv/lib/python3.11/site-packages/litestar/_asgi/routing_trie/__pycache__/__init__.cpython-311.pyc
+++ /dev/null
Binary files differ
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
deleted file mode 100644
index 73323bf..0000000
--- a/venv/lib/python3.11/site-packages/litestar/_asgi/routing_trie/__pycache__/mapping.cpython-311.pyc
+++ /dev/null
Binary files differ
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
deleted file mode 100644
index 9ed2983..0000000
--- a/venv/lib/python3.11/site-packages/litestar/_asgi/routing_trie/__pycache__/traversal.cpython-311.pyc
+++ /dev/null
Binary files differ
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
deleted file mode 100644
index 5247c5b..0000000
--- a/venv/lib/python3.11/site-packages/litestar/_asgi/routing_trie/__pycache__/types.cpython-311.pyc
+++ /dev/null
Binary files differ
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
deleted file mode 100644
index 57fdcdd..0000000
--- a/venv/lib/python3.11/site-packages/litestar/_asgi/routing_trie/__pycache__/validate.cpython-311.pyc
+++ /dev/null
Binary files differ
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