diff options
Diffstat (limited to 'venv/lib/python3.11/site-packages/litestar/controller.py')
| -rw-r--r-- | venv/lib/python3.11/site-packages/litestar/controller.py | 262 |
1 files changed, 262 insertions, 0 deletions
diff --git a/venv/lib/python3.11/site-packages/litestar/controller.py b/venv/lib/python3.11/site-packages/litestar/controller.py new file mode 100644 index 0000000..967454b --- /dev/null +++ b/venv/lib/python3.11/site-packages/litestar/controller.py @@ -0,0 +1,262 @@ +from __future__ import annotations + +import types +from collections import defaultdict +from copy import deepcopy +from operator import attrgetter +from typing import TYPE_CHECKING, Any, Mapping, Sequence, cast + +from litestar._layers.utils import narrow_response_cookies, narrow_response_headers +from litestar.exceptions import ImproperlyConfiguredException +from litestar.handlers.base import BaseRouteHandler +from litestar.handlers.http_handlers import HTTPRouteHandler +from litestar.handlers.websocket_handlers import WebsocketRouteHandler +from litestar.types.empty import Empty +from litestar.utils import ensure_async_callable, normalize_path +from litestar.utils.signature import add_types_to_signature_namespace + +__all__ = ("Controller",) + + +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.router import Router + from litestar.types import ( + AfterRequestHookHandler, + AfterResponseHookHandler, + BeforeRequestHookHandler, + Dependencies, + ExceptionHandlersMap, + Guard, + Middleware, + ParametersMap, + ResponseCookies, + TypeEncodersMap, + ) + from litestar.types.composite_types import ResponseHeaders, TypeDecodersSequence + from litestar.types.empty import EmptyType + + +class Controller: + """The Litestar Controller class. + + Subclass this class to create 'view' like components and utilize OOP. + """ + + __slots__ = ( + "after_request", + "after_response", + "before_request", + "dependencies", + "dto", + "etag", + "exception_handlers", + "guards", + "include_in_schema", + "middleware", + "opt", + "owner", + "parameters", + "path", + "request_class", + "response_class", + "response_cookies", + "response_headers", + "return_dto", + "security", + "signature_namespace", + "tags", + "type_encoders", + "type_decoders", + "websocket_class", + ) + + after_request: AfterRequestHookHandler | None + """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: AfterResponseHookHandler | None + """A sync or async function called after the response has been awaited. + + It receives the :class:`Request <.connection.Request>` instance and should not return any values. + """ + before_request: BeforeRequestHookHandler | None + """A sync or async function called immediately before calling the route handler. + + It receives the :class:`Request <.connection.Request>` instance and any non-``None`` return value is used for the + response, bypassing the route handler. + """ + cache_control: CacheControlHeader | None + """A :class:`CacheControlHeader <.datastructures.CacheControlHeader>` header to add to route handlers of this + controller. + + Can be overridden by route handlers. + """ + dependencies: Dependencies | None + """A string keyed dictionary of dependency :class:`Provider <.di.Provide>` instances.""" + dto: type[AbstractDTO] | None | EmptyType + """:class:`AbstractDTO <.dto.base_dto.AbstractDTO>` to use for (de)serializing and validation of request data.""" + etag: ETag | None + """An ``etag`` header of type :class:`ETag <.datastructures.ETag>` to add to route handlers of this controller. + + Can be overridden by route handlers. + """ + exception_handlers: ExceptionHandlersMap | None + """A map of handler functions to status codes and/or exception types.""" + guards: Sequence[Guard] | None + """A sequence of :class:`Guard <.types.Guard>` callables.""" + include_in_schema: bool | EmptyType + """A boolean flag dictating whether the route handler should be documented in the OpenAPI schema""" + middleware: Sequence[Middleware] | None + """A sequence of :class:`Middleware <.types.Middleware>`.""" + opt: Mapping[str, Any] | None + """A string key mapping of arbitrary values that can be accessed in :class:`Guards <.types.Guard>` or wherever you + have access to :class:`Request <.connection.Request>` or :class:`ASGI Scope <.types.Scope>`. + """ + owner: Router + """The :class:`Router <.router.Router>` or :class:`Litestar <litestar.app.Litestar>` app that owns the controller. + + This value is set internally by Litestar and it should not be set when subclassing the controller. + """ + parameters: ParametersMap | None + """A mapping of :class:`Parameter <.params.Parameter>` definitions available to all application paths.""" + path: str + """A path fragment for the controller. + + All route handlers under the controller will have the fragment appended to them. If not set it defaults to ``/``. + """ + request_class: type[Request] | None + """A custom subclass of :class:`Request <.connection.Request>` to be used as the default request for all route + handlers under the controller. + """ + response_class: type[Response] | None + """A custom subclass of :class:`Response <.response.Response>` to be used as the default response for all route + handlers under the controller. + """ + response_cookies: ResponseCookies | None + """A list of :class:`Cookie <.datastructures.Cookie>` instances.""" + response_headers: ResponseHeaders | None + """A string keyed dictionary mapping :class:`ResponseHeader <.datastructures.ResponseHeader>` instances.""" + return_dto: type[AbstractDTO] | None | EmptyType + """:class:`AbstractDTO <.dto.base_dto.AbstractDTO>` to use for serializing outbound response + data. + """ + tags: Sequence[str] | None + """A sequence of string tags that will be appended to the schema of all route handlers under the controller.""" + security: Sequence[SecurityRequirement] | None + """A sequence of dictionaries that to the schema of all route handlers under the controller.""" + signature_namespace: dict[str, Any] + """A mapping of names to types for use in forward reference resolution during signature modeling.""" + signature_types: Sequence[Any] + """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. + """ + type_decoders: TypeDecodersSequence | None + """A sequence of tuples, each composed of a predicate testing for type identity and a msgspec hook for deserialization.""" + type_encoders: TypeEncodersMap | None + """A mapping of types to callables that transform them into types supported for serialization.""" + websocket_class: type[WebSocket] | None + """A custom subclass of :class:`WebSocket <.connection.WebSocket>` to be used as the default websocket for all route + handlers under the controller. + """ + + def __init__(self, owner: Router) -> None: + """Initialize a controller. + + Should only be called by routers as part of controller registration. + + Args: + owner: An instance of :class:`Router <.router.Router>` + """ + # Since functions set on classes are bound, we need replace the bound instance with the class version and wrap + # it to ensure it does not get bound. + for key in ("after_request", "after_response", "before_request"): + cls_value = getattr(type(self), key, None) + if callable(cls_value): + setattr(self, key, ensure_async_callable(cls_value)) + + if not hasattr(self, "dto"): + self.dto = Empty + + if not hasattr(self, "return_dto"): + self.return_dto = Empty + + if not hasattr(self, "include_in_schema"): + self.include_in_schema = Empty + + self.signature_namespace = add_types_to_signature_namespace( + getattr(self, "signature_types", []), getattr(self, "signature_namespace", {}) + ) + + for key in self.__slots__: + if not hasattr(self, key): + setattr(self, key, None) + + self.response_cookies = narrow_response_cookies(self.response_cookies) + self.response_headers = narrow_response_headers(self.response_headers) + self.path = normalize_path(self.path or "/") + self.owner = owner + + def get_route_handlers(self) -> list[BaseRouteHandler]: + """Get a controller's route handlers and set the controller as the handlers' owner. + + Returns: + A list containing a copy of the route handlers defined on the controller + """ + + route_handlers: list[BaseRouteHandler] = [] + controller_names = set(dir(Controller)) + self_handlers = [ + getattr(self, name) + for name in dir(self) + if name not in controller_names and isinstance(getattr(self, name), BaseRouteHandler) + ] + self_handlers.sort(key=attrgetter("handler_id")) + for self_handler in self_handlers: + route_handler = deepcopy(self_handler) + # at the point we get a reference to the handler function, it's unbound, so + # we replace it with a regular bound method here + route_handler._fn = types.MethodType(route_handler._fn, self) + route_handler.owner = self + route_handlers.append(route_handler) + + self.validate_route_handlers(route_handlers=route_handlers) + + return route_handlers + + def validate_route_handlers(self, route_handlers: list[BaseRouteHandler]) -> None: + """Validate that the combination of path and decorator method or type are unique on the controller. + + Args: + route_handlers: The controller's route handlers. + + Raises: + ImproperlyConfiguredException + + Returns: + None + """ + paths: defaultdict[str, set[str]] = defaultdict(set) + + for route_handler in route_handlers: + if isinstance(route_handler, HTTPRouteHandler): + methods: set[str] = cast("set[str]", route_handler.http_methods) + elif isinstance(route_handler, WebsocketRouteHandler): + methods = {"websocket"} + else: + methods = {"asgi"} + + for path in route_handler.paths: + if (entry := paths[path]) and (intersection := entry.intersection(methods)): + raise ImproperlyConfiguredException( + f"the combination of path and method must be unique in a controller - " + f"the following methods {''.join(m.lower() for m in intersection)} for {type(self).__name__} " + f"controller path {path} are not unique" + ) + paths[path].update(methods) |