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, 0 insertions, 262 deletions
diff --git a/venv/lib/python3.11/site-packages/litestar/controller.py b/venv/lib/python3.11/site-packages/litestar/controller.py deleted file mode 100644 index 967454b..0000000 --- a/venv/lib/python3.11/site-packages/litestar/controller.py +++ /dev/null @@ -1,262 +0,0 @@ -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) |