from __future__ import annotations
import types
from collections import defaultdict
from operator import attrgetter
from typing import TYPE_CHECKING, Any, 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 normalize_path
from litestar.utils.signature import add_types_to_signature_namespace
__all__ = ("Controller",)
if TYPE_CHECKING:
from collections.abc import Mapping, Sequence
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
[docs]
class Controller:
"""The Litestar Controller class.
Subclass this class to create 'view' like components and utilize OOP.
"""
__slots__ = (
"after_request",
"after_response",
"before_request",
"cache_control",
"dependencies",
"dto",
"etag",
"exception_handlers",
"guards",
"include_in_schema",
"middleware",
"opt",
"parameters",
"path",
"request_class",
"request_max_body_size",
"response_class",
"response_cookies",
"response_headers",
"return_dto",
"security",
"signature_namespace",
"signature_types",
"tags",
"type_decoders",
"type_encoders",
"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>`.
"""
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.
"""
request_max_body_size: int | None | EmptyType
"""
Maximum allowed size of the request body in bytes. If this size is exceeded, a '413 - Request Entity Too Large'
error response is returned."""
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 modelling."""
signature_types: Sequence[Any]
"""A sequence of types for use in forward reference resolution during signature modelling.
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.
"""
[docs]
def __init__(self) -> None:
"""Initialize a controller.
Should only be called by routers as part of controller registration.
"""
# Since functions set on classes are bound, we need replace the bound instance with the class version
for key in ("after_request", "after_response", "before_request"):
cls_value = getattr(type(self), key, None)
if callable(cls_value):
setattr(self, key, 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
if not hasattr(self, "request_max_body_size"):
self.request_max_body_size = 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 "/")
def as_router(self) -> Router:
from litestar.router import Router
return Router(
path=self.path,
route_handlers=self.get_route_handlers(), # type: ignore[arg-type]
after_request=self.after_request,
after_response=self.after_response,
before_request=self.before_request,
cache_control=self.cache_control,
dependencies=self.dependencies,
dto=self.dto,
etag=self.etag,
exception_handlers=self.exception_handlers,
guards=self.guards,
include_in_schema=self.include_in_schema,
middleware=self.middleware,
opt=self.opt,
parameters=self.parameters,
request_class=self.request_class,
response_class=self.response_class,
response_cookies=self.response_cookies,
response_headers=self.response_headers,
return_dto=self.return_dto,
security=self.security,
signature_types=self.signature_types,
signature_namespace=self.signature_namespace,
tags=self.tags,
type_encoders=self.type_encoders,
type_decoders=self.type_decoders,
websocket_class=self.websocket_class,
request_max_body_size=self.request_max_body_size,
)
[docs]
def get_route_handlers(self) -> list[BaseRouteHandler]:
"""Get a controller's route handlers
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: list[BaseRouteHandler] = [
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:
# 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_handlers.append(self_handler._with_changes(fn=types.MethodType(self_handler.fn, self)))
self.validate_route_handlers(route_handlers=route_handlers)
return route_handlers
[docs]
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)
Litestar