path: root/venv/lib/python3.11/site-packages/litestar/handlers/http_handlers/base.py

from __future__ import annotations

from enum import Enum
from typing import TYPE_CHECKING, AnyStr, Mapping, Sequence, TypedDict, cast

from litestar._layers.utils import narrow_response_cookies, narrow_response_headers
from litestar.connection import Request
from litestar.datastructures.cookie import Cookie
from litestar.datastructures.response_header import ResponseHeader
from litestar.enums import HttpMethod, MediaType
from litestar.exceptions import (
    HTTPException,
    ImproperlyConfiguredException,
)
from litestar.handlers.base import BaseRouteHandler
from litestar.handlers.http_handlers._utils import (
    create_data_handler,
    create_generic_asgi_response_handler,
    create_response_handler,
    get_default_status_code,
    is_empty_response_annotation,
    normalize_http_method,
)
from litestar.openapi.spec import Operation
from litestar.response import Response
from litestar.status_codes import HTTP_204_NO_CONTENT, HTTP_304_NOT_MODIFIED
from litestar.types import (
    AfterRequestHookHandler,
    AfterResponseHookHandler,
    AnyCallable,
    ASGIApp,
    BeforeRequestHookHandler,
    CacheKeyBuilder,
    Dependencies,
    Empty,
    EmptyType,
    ExceptionHandlersMap,
    Guard,
    Method,
    Middleware,
    ResponseCookies,
    ResponseHeaders,
    TypeEncodersMap,
)
from litestar.utils import ensure_async_callable
from litestar.utils.predicates import is_async_callable
from litestar.utils.warnings import warn_implicit_sync_to_thread, warn_sync_to_thread_with_async_callable

if TYPE_CHECKING:
    from typing import Any, Awaitable, Callable

    from litestar.app import Litestar
    from litestar.background_tasks import BackgroundTask, BackgroundTasks
    from litestar.config.response_cache import CACHE_FOREVER
    from litestar.datastructures import CacheControlHeader, ETag
    from litestar.dto import AbstractDTO
    from litestar.openapi.datastructures import ResponseSpec
    from litestar.openapi.spec import SecurityRequirement
    from litestar.types.callable_types import AsyncAnyCallable, OperationIDCreator
    from litestar.types.composite_types import TypeDecodersSequence

__all__ = ("HTTPRouteHandler", "route")


class ResponseHandlerMap(TypedDict):
    default_handler: Callable[[Any], Awaitable[ASGIApp]] | EmptyType
    response_type_handler: Callable[[Any], Awaitable[ASGIApp]] | EmptyType


class HTTPRouteHandler(BaseRouteHandler):
    """HTTP Route Decorator.

    Use this decorator to decorate an HTTP handler with multiple methods.
    """

    __slots__ = (
        "_resolved_after_response",
        "_resolved_before_request",
        "_response_handler_mapping",
        "_resolved_include_in_schema",
        "_resolved_tags",
        "_resolved_security",
        "after_request",
        "after_response",
        "background",
        "before_request",
        "cache",
        "cache_control",
        "cache_key_builder",
        "content_encoding",
        "content_media_type",
        "deprecated",
        "description",
        "etag",
        "has_sync_callable",
        "http_methods",
        "include_in_schema",
        "media_type",
        "operation_class",
        "operation_id",
        "raises",
        "request_class",
        "response_class",
        "response_cookies",
        "response_description",
        "response_headers",
        "responses",
        "security",
        "status_code",
        "summary",
        "sync_to_thread",
        "tags",
        "template_name",
    )

    has_sync_callable: bool

    def __init__(
        self,
        path: str | Sequence[str] | None = None,
        *,
        after_request: AfterRequestHookHandler | None = None,
        after_response: AfterResponseHookHandler | None = None,
        background: BackgroundTask | BackgroundTasks | None = None,
        before_request: BeforeRequestHookHandler | None = None,
        cache: bool | int | type[CACHE_FOREVER] = False,
        cache_control: CacheControlHeader | None = None,
        cache_key_builder: CacheKeyBuilder | 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,
        http_method: HttpMethod | Method | Sequence[HttpMethod | Method],
        media_type: MediaType | str | None = None,
        middleware: Sequence[Middleware] | None = None,
        name: str | None = None,
        opt: Mapping[str, Any] | 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,
        status_code: int | None = None,
        sync_to_thread: bool | None = None,
        # OpenAPI related attributes
        content_encoding: str | None = None,
        content_media_type: str | None = None,
        deprecated: bool = False,
        description: str | None = None,
        include_in_schema: bool | EmptyType = Empty,
        operation_class: type[Operation] = Operation,
        operation_id: str | OperationIDCreator | None = None,
        raises: Sequence[type[HTTPException]] | None = None,
        response_description: str | None = None,
        responses: Mapping[int, ResponseSpec] | None = None,
        signature_namespace: Mapping[str, Any] | None = None,
        security: Sequence[SecurityRequirement] | None = None,
        summary: str | None = None,
        tags: Sequence[str] | None = None,
        type_decoders: TypeDecodersSequence | None = None,
        type_encoders: TypeEncodersMap | None = None,
        **kwargs: Any,
    ) -> None:
        """Initialize ``HTTPRouteHandler``.

        Args:
            path: A path fragment for the route handler function or a sequence of path fragments.
                If not given defaults to ``/``
            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.
            background: A :class:`BackgroundTask <.background_tasks.BackgroundTask>` instance or
                :class:`BackgroundTasks <.background_tasks.BackgroundTasks>` to execute after the response is finished.
                Defaults to ``None``.
            before_request: A sync or async function called immediately before calling the route handler. Receives
                the :class:`Request <.connection.Request>` instance and any non-``None`` return value is used for the
                response, bypassing the route handler.
            cache: Enables response caching if configured on the application level. Valid values are ``True`` or a
                number of seconds (e.g. ``120``) to cache the response.
            cache_control: A ``cache-control`` header of type
                :class:`CacheControlHeader <.datastructures.CacheControlHeader>` that will be added to the response.
            cache_key_builder: A :class:`cache-key builder function <.types.CacheKeyBuilder>`. Allows for customization
                of the cache key if caching is configured on the application level.
            dependencies: A string keyed mapping of dependency :class:`Provider <.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>` that will be added to the response.
            exception_handlers: A mapping of status codes and/or exception types to handler functions.
            guards: A sequence of :class:`Guard <.types.Guard>` callables.
            http_method: An :class:`http method string <.types.Method>`, a member of the enum
                :class:`HttpMethod <.enums.HttpMethod>` or a list of these that correlates to the methods the route
                handler function should handle.
            media_type: A member of the :class:`MediaType <.enums.MediaType>` enum or a string with a valid IANA
                Media-Type.
            middleware: A sequence of :class:`Middleware <.types.Middleware>`.
            name: A string identifying the route handler.
            opt: A string keyed 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>`.
            request_class: A custom subclass of :class:`Request <.connection.Request>` to be used as route handler's
                default request.
            response_class: A custom subclass of :class:`Response <.response.Response>` to be used as route handler's
                default response.
            response_cookies: A sequence of :class:`Cookie <.datastructures.Cookie>` instances.
            response_headers: A string keyed mapping of :class:`ResponseHeader <.datastructures.ResponseHeader>`
                instances.
            responses: A mapping of additional status codes and a description of their expected content.
                This information will be included in the OpenAPI schema
            return_dto: :class:`AbstractDTO <.dto.base_dto.AbstractDTO>` to use for serializing
                outbound response data.
            signature_namespace: A mapping of names to types for use in forward reference resolution during signature modelling.
            status_code: An http status code for the response. Defaults to ``200`` for mixed method or ``GET``, ``PUT`` and
                ``PATCH``, ``201`` for ``POST`` and ``204`` for ``DELETE``.
            sync_to_thread: A boolean dictating whether the handler function will be executed in a worker thread or the
                main event loop. This has an effect only for sync handler functions. See using sync handler functions.
            content_encoding: A string describing the encoding of the content, e.g. ``"base64"``.
            content_media_type: A string designating the media-type of the content, e.g. ``"image/png"``.
            deprecated:  A boolean dictating whether this route should be marked as deprecated in the OpenAPI schema.
            description: Text used for the route's schema description section.
            include_in_schema: A boolean flag dictating whether  the route handler should be documented in the OpenAPI schema.
            operation_class: :class:`Operation <.openapi.spec.operation.Operation>` to be used with the route's OpenAPI schema.
            operation_id: Either a string or a callable returning a string. An identifier used for the route's schema operationId.
            raises:  A list of exception classes extending from litestar.HttpException that is used for the OpenAPI documentation.
                This list should describe all exceptions raised within the route handler's function/method. The Litestar
                ValidationException will be added automatically for the schema if any validation is involved.
            response_description: Text used for the route's response schema description section.
            security: A sequence of dictionaries that contain information about which security scheme can be used on the endpoint.
            summary: Text used for the route's schema summary section.
            tags: A sequence of string tags that will be appended to the OpenAPI schema.
            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.
            **kwargs: Any additional kwarg - will be set in the opt dictionary.
        """
        if not http_method:
            raise ImproperlyConfiguredException("An http_method kwarg is required")

        self.http_methods = normalize_http_method(http_methods=http_method)
        self.status_code = status_code or get_default_status_code(http_methods=self.http_methods)

        super().__init__(
            path=path,
            dependencies=dependencies,
            dto=dto,
            exception_handlers=exception_handlers,
            guards=guards,
            middleware=middleware,
            name=name,
            opt=opt,
            return_dto=return_dto,
            signature_namespace=signature_namespace,
            type_decoders=type_decoders,
            type_encoders=type_encoders,
            **kwargs,
        )

        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.background = background
        self.before_request = ensure_async_callable(before_request) if before_request else None
        self.cache = cache
        self.cache_control = cache_control
        self.cache_key_builder = cache_key_builder
        self.etag = etag
        self.media_type: MediaType | str = media_type or ""
        self.request_class = request_class
        self.response_class = response_class
        self.response_cookies: Sequence[Cookie] | None = narrow_response_cookies(response_cookies)
        self.response_headers: Sequence[ResponseHeader] | None = narrow_response_headers(response_headers)

        self.sync_to_thread = sync_to_thread
        # OpenAPI related attributes
        self.content_encoding = content_encoding
        self.content_media_type = content_media_type
        self.deprecated = deprecated
        self.description = description
        self.include_in_schema = include_in_schema
        self.operation_class = operation_class
        self.operation_id = operation_id
        self.raises = raises
        self.response_description = response_description
        self.summary = summary
        self.tags = tags
        self.security = security
        self.responses = responses
        # memoized attributes, defaulted to Empty
        self._resolved_after_response: AsyncAnyCallable | None | EmptyType = Empty
        self._resolved_before_request: AsyncAnyCallable | None | EmptyType = Empty
        self._response_handler_mapping: ResponseHandlerMap = {"default_handler": Empty, "response_type_handler": Empty}
        self._resolved_include_in_schema: bool | EmptyType = Empty
        self._resolved_security: list[SecurityRequirement] | EmptyType = Empty
        self._resolved_tags: list[str] | EmptyType = Empty

    def __call__(self, fn: AnyCallable) -> HTTPRouteHandler:
        """Replace a function with itself."""
        if not is_async_callable(fn):
            if self.sync_to_thread is None:
                warn_implicit_sync_to_thread(fn, stacklevel=3)
        elif self.sync_to_thread is not None:
            warn_sync_to_thread_with_async_callable(fn, stacklevel=3)

        super().__call__(fn)
        return self

    def resolve_request_class(self) -> type[Request]:
        """Return the closest custom Request class in the owner graph or the default Request class.

        This method is memoized so the computation occurs only once.

        Returns:
            The default :class:`Request <.connection.Request>` class for the route handler.
        """
        return next(
            (layer.request_class for layer in reversed(self.ownership_layers) if layer.request_class is not None),
            Request,
        )

    def resolve_response_class(self) -> type[Response]:
        """Return the closest custom Response class in the owner graph or the default Response class.

        This method is memoized so the computation occurs only once.

        Returns:
            The default :class:`Response <.response.Response>` class for the route handler.
        """
        return next(
            (layer.response_class for layer in reversed(self.ownership_layers) if layer.response_class is not None),
            Response,
        )

    def resolve_response_headers(self) -> frozenset[ResponseHeader]:
        """Return all header parameters in the scope of the handler function.

        Returns:
            A dictionary mapping keys to :class:`ResponseHeader <.datastructures.ResponseHeader>` instances.
        """
        resolved_response_headers: dict[str, ResponseHeader] = {}

        for layer in self.ownership_layers:
            if layer_response_headers := layer.response_headers:
                if isinstance(layer_response_headers, Mapping):
                    # this can't happen unless you manually set response_headers on an instance, which would result in a
                    # type-checking error on everything but the controller. We cover this case nevertheless
                    resolved_response_headers.update(
                        {name: ResponseHeader(name=name, value=value) for name, value in layer_response_headers.items()}
                    )
                else:
                    resolved_response_headers.update({h.name: h for h in layer_response_headers})
            for extra_header in ("cache_control", "etag"):
                if header_model := getattr(layer, extra_header, None):
                    resolved_response_headers[header_model.HEADER_NAME] = ResponseHeader(
                        name=header_model.HEADER_NAME,
                        value=header_model.to_header(),
                        documentation_only=header_model.documentation_only,
                    )

        return frozenset(resolved_response_headers.values())

    def resolve_response_cookies(self) -> frozenset[Cookie]:
        """Return a list of Cookie instances. Filters the list to ensure each cookie key is unique.

        Returns:
            A list of :class:`Cookie <.datastructures.Cookie>` instances.
        """
        response_cookies: set[Cookie] = set()
        for layer in reversed(self.ownership_layers):
            if layer_response_cookies := layer.response_cookies:
                if isinstance(layer_response_cookies, Mapping):
                    # this can't happen unless you manually set response_cookies on an instance, which would result in a
                    # type-checking error on everything but the controller. We cover this case nevertheless
                    response_cookies.update(
                        {Cookie(key=key, value=value) for key, value in layer_response_cookies.items()}
                    )
                else:
                    response_cookies.update(cast("set[Cookie]", layer_response_cookies))
        return frozenset(response_cookies)

    def resolve_before_request(self) -> AsyncAnyCallable | None:
        """Resolve the before_handler handler by starting from the route handler and moving up.

        If a handler is found it is returned, otherwise None is set.
        This method is memoized so the computation occurs only once.

        Returns:
            An optional :class:`before request lifecycle hook handler <.types.BeforeRequestHookHandler>`
        """
        if self._resolved_before_request is Empty:
            before_request_handlers = [layer.before_request for layer in self.ownership_layers if layer.before_request]
            self._resolved_before_request = before_request_handlers[-1] if before_request_handlers else None
        return cast("AsyncAnyCallable | None", self._resolved_before_request)

    def resolve_after_response(self) -> AsyncAnyCallable | None:
        """Resolve the after_response handler by starting from the route handler and moving up.

        If a handler is found it is returned, otherwise None is set.
        This method is memoized so the computation occurs only once.

        Returns:
            An optional :class:`after response lifecycle hook handler <.types.AfterResponseHookHandler>`
        """
        if self._resolved_after_response is Empty:
            after_response_handlers: list[AsyncAnyCallable] = [
                layer.after_response  # type: ignore[misc]
                for layer in self.ownership_layers
                if layer.after_response
            ]
            self._resolved_after_response = after_response_handlers[-1] if after_response_handlers else None

        return cast("AsyncAnyCallable | None", self._resolved_after_response)

    def resolve_include_in_schema(self) -> bool:
        """Resolve the 'include_in_schema' property by starting from the route handler and moving up.

        If 'include_in_schema' is found in any of the ownership layers, the last value found is returned.
        If not found in any layer, the default value ``True`` is returned.

        Returns:
            bool: The resolved 'include_in_schema' property.
        """
        if self._resolved_include_in_schema is Empty:
            include_in_schemas = [
                i.include_in_schema for i in self.ownership_layers if isinstance(i.include_in_schema, bool)
            ]
            self._resolved_include_in_schema = include_in_schemas[-1] if include_in_schemas else True

        return self._resolved_include_in_schema

    def resolve_security(self) -> list[SecurityRequirement]:
        """Resolve the security property by starting from the route handler and moving up.

        Security requirements are additive, so the security requirements of the route handler are the sum of all
        security requirements of the ownership layers.

        Returns:
            list[SecurityRequirement]: The resolved security property.
        """
        if self._resolved_security is Empty:
            self._resolved_security = []
            for layer in self.ownership_layers:
                if isinstance(layer.security, Sequence):
                    self._resolved_security.extend(layer.security)

        return self._resolved_security

    def resolve_tags(self) -> list[str]:
        """Resolve the tags property by starting from the route handler and moving up.

        Tags are additive, so the tags of the route handler are the sum of all tags of the ownership layers.

        Returns:
            list[str]: A sorted list of unique tags.
        """
        if self._resolved_tags is Empty:
            tag_set = set()
            for layer in self.ownership_layers:
                for tag in layer.tags or []:
                    tag_set.add(tag)
            self._resolved_tags = sorted(tag_set)

        return self._resolved_tags

    def get_response_handler(self, is_response_type_data: bool = False) -> Callable[[Any], Awaitable[ASGIApp]]:
        """Resolve the response_handler function for the route handler.

        This method is memoized so the computation occurs only once.

        Args:
            is_response_type_data: Whether to return a handler for 'Response' instances.

        Returns:
            Async Callable to handle an HTTP Request
        """
        if self._response_handler_mapping["default_handler"] is Empty:
            after_request_handlers: list[AsyncAnyCallable] = [
                layer.after_request  # type: ignore[misc]
                for layer in self.ownership_layers
                if layer.after_request
            ]
            after_request = cast(
                "AfterRequestHookHandler | None",
                after_request_handlers[-1] if after_request_handlers else None,
            )

            media_type = self.media_type.value if isinstance(self.media_type, Enum) else self.media_type
            response_class = self.resolve_response_class()
            headers = self.resolve_response_headers()
            cookies = self.resolve_response_cookies()
            type_encoders = self.resolve_type_encoders()

            return_type = self.parsed_fn_signature.return_type
            return_annotation = return_type.annotation

            self._response_handler_mapping["response_type_handler"] = response_type_handler = create_response_handler(
                after_request=after_request,
                background=self.background,
                cookies=cookies,
                headers=headers,
                media_type=media_type,
                status_code=self.status_code,
                type_encoders=type_encoders,
            )

            if return_type.is_subclass_of(Response):
                self._response_handler_mapping["default_handler"] = response_type_handler
            elif is_async_callable(return_annotation) or return_annotation is ASGIApp:
                self._response_handler_mapping["default_handler"] = create_generic_asgi_response_handler(
                    after_request=after_request
                )
            else:
                self._response_handler_mapping["default_handler"] = create_data_handler(
                    after_request=after_request,
                    background=self.background,
                    cookies=cookies,
                    headers=headers,
                    media_type=media_type,
                    response_class=response_class,
                    status_code=self.status_code,
                    type_encoders=type_encoders,
                )

        return cast(
            "Callable[[Any], Awaitable[ASGIApp]]",
            self._response_handler_mapping["response_type_handler"]
            if is_response_type_data
            else self._response_handler_mapping["default_handler"],
        )

    async def to_response(self, app: Litestar, data: Any, request: Request) -> ASGIApp:
        """Return a :class:`Response <.response.Response>` from the handler by resolving and calling it.

        Args:
            app: The :class:`Litestar <litestar.app.Litestar>` app instance
            data: Either an instance of a :class:`Response <.response.Response>`,
                a Response instance or an arbitrary value.
            request: A :class:`Request <.connection.Request>` instance

        Returns:
            A Response instance
        """
        if return_dto_type := self.resolve_return_dto():
            data = return_dto_type(request).data_to_encodable_type(data)

        response_handler = self.get_response_handler(is_response_type_data=isinstance(data, Response))
        return await response_handler(app=app, data=data, request=request)  # type: ignore[call-arg]

    def on_registration(self, app: Litestar) -> None:
        super().on_registration(app)
        self.resolve_after_response()
        self.resolve_include_in_schema()
        self.has_sync_callable = not is_async_callable(self.fn)

        if self.has_sync_callable and self.sync_to_thread:
            self._fn = ensure_async_callable(self.fn)
            self.has_sync_callable = False

    def _validate_handler_function(self) -> None:
        """Validate the route handler function once it is set by inspecting its return annotations."""
        super()._validate_handler_function()

        return_type = self.parsed_fn_signature.return_type

        if return_type.annotation is Empty:
            raise ImproperlyConfiguredException(
                "A return value of a route handler function should be type annotated. "
                "If your function doesn't return a value, annotate it as returning 'None'."
            )

        if (
            self.status_code < 200 or self.status_code in {HTTP_204_NO_CONTENT, HTTP_304_NOT_MODIFIED}
        ) and not is_empty_response_annotation(return_type):
            raise ImproperlyConfiguredException(
                "A status code 204, 304 or in the range below 200 does not support a response body. "
                "If the function should return a value, change the route handler status code to an appropriate value.",
            )

        if not self.media_type:
            if return_type.is_subclass_of((str, bytes)) or return_type.annotation is AnyStr:
                self.media_type = MediaType.TEXT
            elif not return_type.is_subclass_of(Response):
                self.media_type = MediaType.JSON

        if "socket" in self.parsed_fn_signature.parameters:
            raise ImproperlyConfiguredException("The 'socket' kwarg is not supported with http handlers")

        if "data" in self.parsed_fn_signature.parameters and "GET" in self.http_methods:
            raise ImproperlyConfiguredException("'data' kwarg is unsupported for 'GET' request handlers")


route = HTTPRouteHandler