summaryrefslogtreecommitdiff
path: root/venv/lib/python3.11/site-packages/litestar/router.py
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/router.py
parentc45662ff3923b34614ddcc8feb9195541166dcc5 (diff)
no venv
Diffstat (limited to 'venv/lib/python3.11/site-packages/litestar/router.py')
-rw-r--r--venv/lib/python3.11/site-packages/litestar/router.py338
1 files changed, 0 insertions, 338 deletions
diff --git a/venv/lib/python3.11/site-packages/litestar/router.py b/venv/lib/python3.11/site-packages/litestar/router.py
deleted file mode 100644
index 85346d8..0000000
--- a/venv/lib/python3.11/site-packages/litestar/router.py
+++ /dev/null
@@ -1,338 +0,0 @@
-from __future__ import annotations
-
-from collections import defaultdict
-from copy import copy, deepcopy
-from typing import TYPE_CHECKING, Any, Mapping, Sequence, cast
-
-from litestar._layers.utils import narrow_response_cookies, narrow_response_headers
-from litestar.controller import Controller
-from litestar.exceptions import ImproperlyConfiguredException
-from litestar.handlers.asgi_handlers import ASGIRouteHandler
-from litestar.handlers.http_handlers import HTTPRouteHandler
-from litestar.handlers.websocket_handlers import WebsocketListener, WebsocketRouteHandler
-from litestar.routes import ASGIRoute, HTTPRoute, WebSocketRoute
-from litestar.types.empty import Empty
-from litestar.utils import find_index, is_class_and_subclass, join_paths, normalize_path, unique
-from litestar.utils.signature import add_types_to_signature_namespace
-from litestar.utils.sync import ensure_async_callable
-
-__all__ = ("Router",)
-
-
-if TYPE_CHECKING:
- from litestar.connection import Request, WebSocket
- from litestar.datastructures import CacheControlHeader, ETag
- from litestar.dto import AbstractDTO
- from litestar.openapi.spec import SecurityRequirement
- from litestar.response import Response
- from litestar.routes import BaseRoute
- from litestar.types import (
- AfterRequestHookHandler,
- AfterResponseHookHandler,
- BeforeRequestHookHandler,
- ControllerRouterHandler,
- ExceptionHandlersMap,
- Guard,
- Middleware,
- ParametersMap,
- ResponseCookies,
- RouteHandlerMapItem,
- RouteHandlerType,
- TypeEncodersMap,
- )
- from litestar.types.composite_types import Dependencies, ResponseHeaders, TypeDecodersSequence
- from litestar.types.empty import EmptyType
-
-
-class Router:
- """The Litestar Router class.
-
- A Router instance is used to group controller, routers and route handler functions under a shared path fragment
- """
-
- __slots__ = (
- "after_request",
- "after_response",
- "before_request",
- "cache_control",
- "dependencies",
- "dto",
- "etag",
- "exception_handlers",
- "guards",
- "include_in_schema",
- "middleware",
- "opt",
- "owner",
- "parameters",
- "path",
- "registered_route_handler_ids",
- "request_class",
- "response_class",
- "response_cookies",
- "response_headers",
- "return_dto",
- "routes",
- "security",
- "signature_namespace",
- "tags",
- "type_decoders",
- "type_encoders",
- "websocket_class",
- )
-
- def __init__(
- self,
- path: str,
- *,
- after_request: AfterRequestHookHandler | None = None,
- after_response: AfterResponseHookHandler | None = None,
- before_request: BeforeRequestHookHandler | None = None,
- cache_control: CacheControlHeader | None = None,
- dependencies: Dependencies | None = None,
- dto: type[AbstractDTO] | None | EmptyType = Empty,
- etag: ETag | None = None,
- exception_handlers: ExceptionHandlersMap | None = None,
- guards: Sequence[Guard] | None = None,
- include_in_schema: bool | EmptyType = Empty,
- middleware: Sequence[Middleware] | None = None,
- opt: Mapping[str, Any] | None = None,
- parameters: ParametersMap | None = None,
- request_class: type[Request] | None = None,
- response_class: type[Response] | None = None,
- response_cookies: ResponseCookies | None = None,
- response_headers: ResponseHeaders | None = None,
- return_dto: type[AbstractDTO] | None | EmptyType = Empty,
- route_handlers: Sequence[ControllerRouterHandler],
- security: Sequence[SecurityRequirement] | None = None,
- signature_namespace: Mapping[str, Any] | None = None,
- signature_types: Sequence[Any] | None = None,
- tags: Sequence[str] | None = None,
- type_decoders: TypeDecodersSequence | None = None,
- type_encoders: TypeEncodersMap | None = None,
- websocket_class: type[WebSocket] | None = None,
- ) -> None:
- """Initialize a ``Router``.
-
- Args:
- after_request: A sync or async function executed before a :class:`Request <.connection.Request>` is passed
- to any route handler. If this function returns a value, the request will not reach the route handler,
- and instead this value will be used.
- after_response: A sync or async function called after the response has been awaited. It receives the
- :class:`Request <.connection.Request>` object and should not return any values.
- before_request: A sync or async function called immediately before calling the route handler. Receives
- the :class:`litestar.connection.Request` instance and any non-``None`` return value is used for the
- response, bypassing the route handler.
- cache_control: A ``cache-control`` header of type
- :class:`CacheControlHeader <.datastructures.CacheControlHeader>` to add to route handlers of
- this router. Can be overridden by route handlers.
- dependencies: A string keyed mapping of dependency :class:`Provide <.di.Provide>` instances.
- dto: :class:`AbstractDTO <.dto.base_dto.AbstractDTO>` to use for (de)serializing and
- validation of request data.
- etag: An ``etag`` header of type :class:`ETag <.datastructures.ETag>` to add to route handlers of this app.
- exception_handlers: A mapping of status codes and/or exception types to handler functions.
- guards: A sequence of :data:`Guard <.types.Guard>` callables.
- include_in_schema: A boolean flag dictating whether the route handler should be documented in the OpenAPI schema.
- middleware: A sequence of :data:`Middleware <.types.Middleware>`.
- opt: A string keyed mapping of arbitrary values that can be accessed in :data:`Guards <.types.Guard>` or
- wherever you have access to :class:`Request <.connection.Request>` or
- :data:`ASGI Scope <.types.Scope>`.
- parameters: A mapping of :func:`Parameter <.params.Parameter>` definitions available to all application
- paths.
- path: A path fragment that is prefixed to all route handlers, controllers and other routers associated
- with the router instance.
- request_class: A custom subclass of :class:`Request <.connection.Request>` to be used as the default for
- all route handlers, controllers and other routers associated with the router instance.
- response_class: A custom subclass of :class:`Response <.response.Response>` to be used as the default for
- all route handlers, controllers and other routers associated with the router instance.
- response_cookies: A sequence of :class:`Cookie <.datastructures.Cookie>` instances.
- response_headers: A string keyed mapping of :class:`ResponseHeader <.datastructures.ResponseHeader>`
- instances.
- return_dto: :class:`AbstractDTO <.dto.base_dto.AbstractDTO>` to use for serializing
- outbound response data.
- route_handlers: A required sequence of route handlers, which can include instances of
- :class:`Router <.router.Router>`, subclasses of :class:`Controller <.controller.Controller>` or any
- function decorated by the route handler decorators.
- security: A sequence of dicts that will be added to the schema of all route handlers in the application.
- See :data:`SecurityRequirement <.openapi.spec.SecurityRequirement>`
- for details.
- signature_namespace: A mapping of names to types for use in forward reference resolution during signature modeling.
- signature_types: A sequence of types for use in forward reference resolution during signature modeling.
- These types will be added to the signature namespace using their ``__name__`` attribute.
- tags: A sequence of string tags that will be appended to the schema of all route handlers under the
- application.
- type_decoders: A sequence of tuples, each composed of a predicate testing for type identity and a msgspec hook for deserialization.
- type_encoders: A mapping of types to callables that transform them into types supported for serialization.
- websocket_class: A custom subclass of :class:`WebSocket <.connection.WebSocket>` to be used as the default for
- all route handlers, controllers and other routers associated with the router instance.
- """
-
- self.after_request = ensure_async_callable(after_request) if after_request else None # pyright: ignore
- self.after_response = ensure_async_callable(after_response) if after_response else None
- self.before_request = ensure_async_callable(before_request) if before_request else None
- self.cache_control = cache_control
- self.dto = dto
- self.etag = etag
- self.dependencies = dict(dependencies or {})
- self.exception_handlers = dict(exception_handlers or {})
- self.guards = list(guards or [])
- self.include_in_schema = include_in_schema
- self.middleware = list(middleware or [])
- self.opt = dict(opt or {})
- self.owner: Router | None = None
- self.parameters = dict(parameters or {})
- self.path = normalize_path(path)
- self.request_class = request_class
- self.response_class = response_class
- self.response_cookies = narrow_response_cookies(response_cookies)
- self.response_headers = narrow_response_headers(response_headers)
- self.return_dto = return_dto
- self.routes: list[HTTPRoute | ASGIRoute | WebSocketRoute] = []
- self.security = list(security or [])
- self.signature_namespace = add_types_to_signature_namespace(
- signature_types or [], dict(signature_namespace or {})
- )
- self.tags = list(tags or [])
- self.registered_route_handler_ids: set[int] = set()
- self.type_encoders = dict(type_encoders) if type_encoders is not None else None
- self.type_decoders = list(type_decoders) if type_decoders is not None else None
- self.websocket_class = websocket_class
-
- for route_handler in route_handlers or []:
- self.register(value=route_handler)
-
- def register(self, value: ControllerRouterHandler) -> list[BaseRoute]:
- """Register a Controller, Route instance or RouteHandler on the router.
-
- Args:
- value: a subclass or instance of Controller, an instance of :class:`Router` or a function/method that has
- been decorated by any of the routing decorators, e.g. :class:`get <.handlers.get>`,
- :class:`post <.handlers.post>`.
-
- Returns:
- Collection of handlers added to the router.
- """
- validated_value = self._validate_registration_value(value)
-
- routes: list[BaseRoute] = []
-
- for route_path, handlers_map in self.get_route_handler_map(value=validated_value).items():
- path = join_paths([self.path, route_path])
- if http_handlers := unique(
- [handler for handler in handlers_map.values() if isinstance(handler, HTTPRouteHandler)]
- ):
- if existing_handlers := unique(
- [
- handler
- for handler in self.route_handler_method_map.get(path, {}).values()
- if isinstance(handler, HTTPRouteHandler)
- ]
- ):
- http_handlers.extend(existing_handlers)
- existing_route_index = find_index(self.routes, lambda x: x.path == path) # noqa: B023
-
- if existing_route_index == -1: # pragma: no cover
- raise ImproperlyConfiguredException("unable to find_index existing route index")
-
- route: WebSocketRoute | ASGIRoute | HTTPRoute = HTTPRoute(
- path=path,
- route_handlers=http_handlers,
- )
- self.routes[existing_route_index] = route
- else:
- route = HTTPRoute(path=path, route_handlers=http_handlers)
- self.routes.append(route)
-
- routes.append(route)
-
- if websocket_handler := handlers_map.get("websocket"):
- route = WebSocketRoute(path=path, route_handler=cast("WebsocketRouteHandler", websocket_handler))
- self.routes.append(route)
- routes.append(route)
-
- if asgi_handler := handlers_map.get("asgi"):
- route = ASGIRoute(path=path, route_handler=cast("ASGIRouteHandler", asgi_handler))
- self.routes.append(route)
- routes.append(route)
-
- return routes
-
- @property
- def route_handler_method_map(self) -> dict[str, RouteHandlerMapItem]:
- """Map route paths to :class:`RouteHandlerMapItem <litestar.types.internal_typ es.RouteHandlerMapItem>`
-
- Returns:
- A dictionary mapping paths to route handlers
- """
- route_map: dict[str, RouteHandlerMapItem] = defaultdict(dict)
- for route in self.routes:
- if isinstance(route, HTTPRoute):
- for route_handler in route.route_handlers:
- for method in route_handler.http_methods:
- route_map[route.path][method] = route_handler
- else:
- route_map[route.path]["websocket" if isinstance(route, WebSocketRoute) else "asgi"] = (
- route.route_handler
- )
-
- return route_map
-
- @classmethod
- def get_route_handler_map(
- cls,
- value: Controller | RouteHandlerType | Router,
- ) -> dict[str, RouteHandlerMapItem]:
- """Map route handlers to HTTP methods."""
- if isinstance(value, Router):
- return value.route_handler_method_map
-
- if isinstance(value, (HTTPRouteHandler, ASGIRouteHandler, WebsocketRouteHandler)):
- copied_value = copy(value)
- if isinstance(value, HTTPRouteHandler):
- return {path: {http_method: copied_value for http_method in value.http_methods} for path in value.paths}
-
- return {
- path: {"websocket" if isinstance(value, WebsocketRouteHandler) else "asgi": copied_value}
- for path in value.paths
- }
-
- handlers_map: defaultdict[str, RouteHandlerMapItem] = defaultdict(dict)
- for route_handler in value.get_route_handlers():
- for handler_path in route_handler.paths:
- path = join_paths([value.path, handler_path]) if handler_path else value.path
- if isinstance(route_handler, HTTPRouteHandler):
- for http_method in route_handler.http_methods:
- handlers_map[path][http_method] = route_handler
- else:
- handlers_map[path]["websocket" if isinstance(route_handler, WebsocketRouteHandler) else "asgi"] = (
- cast("WebsocketRouteHandler | ASGIRouteHandler", route_handler)
- )
-
- return handlers_map
-
- def _validate_registration_value(self, value: ControllerRouterHandler) -> Controller | RouteHandlerType | Router:
- """Ensure values passed to the register method are supported."""
- if is_class_and_subclass(value, Controller):
- return value(owner=self)
-
- # this narrows down to an ABC, but we assume a non-abstract subclass of the ABC superclass
- if is_class_and_subclass(value, WebsocketListener):
- return value(owner=self).to_handler() # pyright: ignore
-
- if isinstance(value, Router):
- if value is self:
- raise ImproperlyConfiguredException("Cannot register a router on itself")
-
- router_copy = deepcopy(value)
- router_copy.owner = self
- return router_copy
-
- if isinstance(value, (ASGIRouteHandler, HTTPRouteHandler, WebsocketRouteHandler)):
- value.owner = self
- return value
-
- raise ImproperlyConfiguredException(
- "Unsupported value passed to `Router.register`. "
- "If you passed in a function or method, "
- "make sure to decorate it first with one of the routing decorators"
- )