diff options
Diffstat (limited to 'venv/lib/python3.11/site-packages/litestar/app.py')
-rw-r--r-- | venv/lib/python3.11/site-packages/litestar/app.py | 880 |
1 files changed, 880 insertions, 0 deletions
diff --git a/venv/lib/python3.11/site-packages/litestar/app.py b/venv/lib/python3.11/site-packages/litestar/app.py new file mode 100644 index 0000000..e1bd989 --- /dev/null +++ b/venv/lib/python3.11/site-packages/litestar/app.py @@ -0,0 +1,880 @@ +from __future__ import annotations + +import inspect +import logging +import os +from contextlib import ( + AbstractAsyncContextManager, + AsyncExitStack, + asynccontextmanager, + suppress, +) +from datetime import date, datetime, time, timedelta +from functools import partial +from itertools import chain +from pathlib import Path +from typing import TYPE_CHECKING, Any, AsyncGenerator, Callable, Iterable, Mapping, Sequence, TypedDict, cast + +from litestar._asgi import ASGIRouter +from litestar._asgi.utils import get_route_handlers, wrap_in_exception_handler +from litestar._openapi.plugin import OpenAPIPlugin +from litestar._openapi.schema_generation import openapi_schema_plugins +from litestar.config.allowed_hosts import AllowedHostsConfig +from litestar.config.app import AppConfig +from litestar.config.response_cache import ResponseCacheConfig +from litestar.connection import Request, WebSocket +from litestar.datastructures.state import State +from litestar.events.emitter import BaseEventEmitterBackend, SimpleEventEmitter +from litestar.exceptions import ( + MissingDependencyException, + NoRouteMatchFoundException, +) +from litestar.logging.config import LoggingConfig, get_logger_placeholder +from litestar.middleware.cors import CORSMiddleware +from litestar.openapi.config import OpenAPIConfig +from litestar.plugins import ( + CLIPluginProtocol, + InitPluginProtocol, + OpenAPISchemaPluginProtocol, + PluginProtocol, + PluginRegistry, + SerializationPluginProtocol, +) +from litestar.plugins.base import CLIPlugin +from litestar.router import Router +from litestar.routes import ASGIRoute, HTTPRoute, WebSocketRoute +from litestar.static_files.base import StaticFiles +from litestar.stores.registry import StoreRegistry +from litestar.types import Empty, TypeDecodersSequence +from litestar.types.internal_types import PathParameterDefinition, TemplateConfigType +from litestar.utils import deprecated, ensure_async_callable, join_paths, unique +from litestar.utils.dataclass import extract_dataclass_items +from litestar.utils.predicates import is_async_callable +from litestar.utils.warnings import warn_pdb_on_exception + +if TYPE_CHECKING: + from typing_extensions import Self + + from litestar.config.app import ExperimentalFeatures + from litestar.config.compression import CompressionConfig + from litestar.config.cors import CORSConfig + from litestar.config.csrf import CSRFConfig + from litestar.datastructures import CacheControlHeader, ETag + from litestar.dto import AbstractDTO + from litestar.events.listener import EventListener + from litestar.logging.config import BaseLoggingConfig + from litestar.openapi.spec import SecurityRequirement + from litestar.openapi.spec.open_api import OpenAPI + from litestar.response import Response + from litestar.static_files.config import StaticFilesConfig + from litestar.stores.base import Store + from litestar.types import ( + AfterExceptionHookHandler, + AfterRequestHookHandler, + AfterResponseHookHandler, + AnyCallable, + ASGIApp, + BeforeMessageSendHookHandler, + BeforeRequestHookHandler, + ControllerRouterHandler, + Dependencies, + EmptyType, + ExceptionHandlersMap, + GetLogger, + Guard, + LifeSpanReceive, + LifeSpanScope, + LifeSpanSend, + Logger, + Message, + Middleware, + OnAppInitHandler, + ParametersMap, + Receive, + ResponseCookies, + ResponseHeaders, + RouteHandlerType, + Scope, + Send, + TypeEncodersMap, + ) + from litestar.types.callable_types import LifespanHook + + +__all__ = ("HandlerIndex", "Litestar", "DEFAULT_OPENAPI_CONFIG") + +DEFAULT_OPENAPI_CONFIG = OpenAPIConfig(title="Litestar API", version="1.0.0") +"""The default OpenAPI config used if not configuration is explicitly passed to the +:class:`Litestar <.app.Litestar>` instance constructor. +""" + + +class HandlerIndex(TypedDict): + """Map route handler names to a mapping of paths + route handler. + + It's returned from the 'get_handler_index_by_name' utility method. + """ + + paths: list[str] + """Full route paths to the route handler.""" + handler: RouteHandlerType + """Route handler instance.""" + identifier: str + """Unique identifier of the handler. + + Either equal to :attr`__name__ <obj.__name__>` attribute or ``__str__`` value of the handler. + """ + + +class Litestar(Router): + """The Litestar application. + + ``Litestar`` is the root level of the app - it has the base path of ``/`` and all root level Controllers, Routers + and Route Handlers should be registered on it. + """ + + __slots__ = ( + "_lifespan_managers", + "_server_lifespan_managers", + "_debug", + "_openapi_schema", + "_static_files_config", + "plugins", + "after_exception", + "allowed_hosts", + "asgi_handler", + "asgi_router", + "before_send", + "compression_config", + "cors_config", + "csrf_config", + "event_emitter", + "get_logger", + "include_in_schema", + "logger", + "logging_config", + "multipart_form_part_limit", + "on_shutdown", + "on_startup", + "openapi_config", + "request_class", + "response_cache_config", + "route_map", + "signature_namespace", + "state", + "stores", + "template_engine", + "websocket_class", + "pdb_on_exception", + "experimental_features", + ) + + def __init__( + self, + route_handlers: Sequence[ControllerRouterHandler] | None = None, + *, + after_exception: Sequence[AfterExceptionHookHandler] | None = None, + after_request: AfterRequestHookHandler | None = None, + after_response: AfterResponseHookHandler | None = None, + allowed_hosts: Sequence[str] | AllowedHostsConfig | None = None, + before_request: BeforeRequestHookHandler | None = None, + before_send: Sequence[BeforeMessageSendHookHandler] | None = None, + cache_control: CacheControlHeader | None = None, + compression_config: CompressionConfig | None = None, + cors_config: CORSConfig | None = None, + csrf_config: CSRFConfig | None = None, + dto: type[AbstractDTO] | None | EmptyType = Empty, + debug: bool | None = None, + dependencies: Dependencies | None = None, + etag: ETag | None = None, + event_emitter_backend: type[BaseEventEmitterBackend] = SimpleEventEmitter, + exception_handlers: ExceptionHandlersMap | None = None, + guards: Sequence[Guard] | None = None, + include_in_schema: bool | EmptyType = Empty, + listeners: Sequence[EventListener] | None = None, + logging_config: BaseLoggingConfig | EmptyType | None = Empty, + middleware: Sequence[Middleware] | None = None, + multipart_form_part_limit: int = 1000, + on_app_init: Sequence[OnAppInitHandler] | None = None, + on_shutdown: Sequence[LifespanHook] | None = None, + on_startup: Sequence[LifespanHook] | None = None, + openapi_config: OpenAPIConfig | None = DEFAULT_OPENAPI_CONFIG, + opt: Mapping[str, Any] | None = None, + parameters: ParametersMap | None = None, + plugins: Sequence[PluginProtocol] | None = None, + request_class: type[Request] | None = None, + response_cache_config: ResponseCacheConfig | 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, + security: Sequence[SecurityRequirement] | None = None, + signature_namespace: Mapping[str, Any] | None = None, + signature_types: Sequence[Any] | None = None, + state: State | None = None, + static_files_config: Sequence[StaticFilesConfig] | None = None, + stores: StoreRegistry | dict[str, Store] | None = None, + tags: Sequence[str] | None = None, + template_config: TemplateConfigType | None = None, + type_decoders: TypeDecodersSequence | None = None, + type_encoders: TypeEncodersMap | None = None, + websocket_class: type[WebSocket] | None = None, + lifespan: Sequence[Callable[[Litestar], AbstractAsyncContextManager] | AbstractAsyncContextManager] + | None = None, + pdb_on_exception: bool | None = None, + experimental_features: Iterable[ExperimentalFeatures] | None = None, + ) -> None: + """Initialize a ``Litestar`` application. + + Args: + after_exception: A sequence of :class:`exception hook handlers <.types.AfterExceptionHookHandler>`. This + hook is called after an exception occurs. In difference to exception handlers, it is not meant to + return a response - only to process the exception (e.g. log it, send it to Sentry etc.). + after_request: A sync or async function executed after the route handler function returned and the response + object has been resolved. Receives the response object. + 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. + allowed_hosts: A sequence of allowed hosts, or an + :class:`AllowedHostsConfig <.config.allowed_hosts.AllowedHostsConfig>` instance. Enables the builtin + allowed hosts middleware. + 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. + before_send: A sequence of :class:`before send hook handlers <.types.BeforeMessageSendHookHandler>`. Called + when the ASGI send function is called. + cache_control: A ``cache-control`` header of type + :class:`CacheControlHeader <litestar.datastructures.CacheControlHeader>` to add to route handlers of + this app. Can be overridden by route handlers. + compression_config: Configures compression behaviour of the application, this enabled a builtin or user + defined Compression middleware. + cors_config: If set, configures :class:`CORSMiddleware <.middleware.cors.CORSMiddleware>`. + csrf_config: If set, configures :class:`CSRFMiddleware <.middleware.csrf.CSRFMiddleware>`. + debug: If ``True``, app errors rendered as HTML with a stack trace. + dependencies: A string keyed mapping of dependency :class:`Providers <.di.Provide>`. + 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. + Can be overridden by route handlers. + event_emitter_backend: A subclass of + :class:`BaseEventEmitterBackend <.events.emitter.BaseEventEmitterBackend>`. + exception_handlers: A mapping of status codes and/or exception types to handler functions. + guards: A sequence of :class:`Guard <.types.Guard>` callables. + include_in_schema: A boolean flag dictating whether the route handler should be documented in the OpenAPI schema. + lifespan: A list of callables returning async context managers, wrapping the lifespan of the ASGI application + listeners: A sequence of :class:`EventListener <.events.listener.EventListener>`. + logging_config: A subclass of :class:`BaseLoggingConfig <.logging.config.BaseLoggingConfig>`. + middleware: A sequence of :class:`Middleware <.types.Middleware>`. + multipart_form_part_limit: The maximal number of allowed parts in a multipart/formdata request. This limit + is intended to protect from DoS attacks. + on_app_init: A sequence of :class:`OnAppInitHandler <.types.OnAppInitHandler>` instances. Handlers receive + an instance of :class:`AppConfig <.config.app.AppConfig>` that will have been initially populated with + the parameters passed to :class:`Litestar <litestar.app.Litestar>`, and must return an instance of same. + If more than one handler is registered they are called in the order they are provided. + on_shutdown: A sequence of :class:`LifespanHook <.types.LifespanHook>` called during application + shutdown. + on_startup: A sequence of :class:`LifespanHook <litestar.types.LifespanHook>` called during + application startup. + openapi_config: Defaults to :attr:`DEFAULT_OPENAPI_CONFIG` + 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 <litestar.connection.request.Request>` or + :class:`ASGI Scope <.types.Scope>`. + parameters: A mapping of :class:`Parameter <.params.Parameter>` definitions available to all application + paths. + pdb_on_exception: Drop into the PDB when an exception occurs. + plugins: Sequence of plugins. + request_class: An optional subclass of :class:`Request <.connection.Request>` to use for http connections. + response_class: A custom subclass of :class:`Response <.response.Response>` to be used as the app's default + response. + response_cookies: A sequence of :class:`Cookie <.datastructures.Cookie>`. + response_headers: A string keyed mapping of :class:`ResponseHeader <.datastructures.ResponseHeader>` + response_cache_config: Configures caching behavior of the application. + return_dto: :class:`AbstractDTO <.dto.base_dto.AbstractDTO>` to use for serializing + outbound response data. + route_handlers: A sequence of route handlers, which can include instances of + :class:`Router <.router.Router>`, subclasses of :class:`Controller <.controller.Controller>` or any + callable 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. + state: An optional :class:`State <.datastructures.State>` for application state. + static_files_config: A sequence of :class:`StaticFilesConfig <.static_files.StaticFilesConfig>` + stores: Central registry of :class:`Store <.stores.base.Store>` that will be available throughout the + application. If this is a dictionary to it will be passed to a + :class:`StoreRegistry <.stores.registry.StoreRegistry>`. If it is a + :class:`StoreRegistry <.stores.registry.StoreRegistry>`, this instance will be used directly. + tags: A sequence of string tags that will be appended to the schema of all route handlers under the + application. + template_config: An instance of :class:`TemplateConfig <.template.TemplateConfig>` + 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: An optional subclass of :class:`WebSocket <.connection.WebSocket>` to use for websocket + connections. + experimental_features: An iterable of experimental features to enable + """ + + if logging_config is Empty: + logging_config = LoggingConfig() + + if debug is None: + debug = os.getenv("LITESTAR_DEBUG", "0") == "1" + + if pdb_on_exception is None: + pdb_on_exception = os.getenv("LITESTAR_PDB", "0") == "1" + + config = AppConfig( + after_exception=list(after_exception or []), + after_request=after_request, + after_response=after_response, + allowed_hosts=allowed_hosts if isinstance(allowed_hosts, AllowedHostsConfig) else list(allowed_hosts or []), + before_request=before_request, + before_send=list(before_send or []), + cache_control=cache_control, + compression_config=compression_config, + cors_config=cors_config, + csrf_config=csrf_config, + debug=debug, + dependencies=dict(dependencies or {}), + dto=dto, + etag=etag, + event_emitter_backend=event_emitter_backend, + exception_handlers=exception_handlers or {}, + guards=list(guards or []), + include_in_schema=include_in_schema, + lifespan=list(lifespan or []), + listeners=list(listeners or []), + logging_config=logging_config, + middleware=list(middleware or []), + multipart_form_part_limit=multipart_form_part_limit, + on_shutdown=list(on_shutdown or []), + on_startup=list(on_startup or []), + openapi_config=openapi_config, + opt=dict(opt or {}), + parameters=parameters or {}, + pdb_on_exception=pdb_on_exception, + plugins=self._get_default_plugins(list(plugins or [])), + request_class=request_class, + response_cache_config=response_cache_config or ResponseCacheConfig(), + response_class=response_class, + response_cookies=response_cookies or [], + response_headers=response_headers or [], + return_dto=return_dto, + route_handlers=list(route_handlers) if route_handlers is not None else [], + security=list(security or []), + signature_namespace=dict(signature_namespace or {}), + signature_types=list(signature_types or []), + state=state or State(), + static_files_config=list(static_files_config or []), + stores=stores, + tags=list(tags or []), + template_config=template_config, + type_encoders=type_encoders, + type_decoders=type_decoders, + websocket_class=websocket_class, + experimental_features=list(experimental_features or []), + ) + + config.plugins.extend([OpenAPIPlugin(self), *openapi_schema_plugins]) + + for handler in chain( + on_app_init or [], + (p.on_app_init for p in config.plugins if isinstance(p, InitPluginProtocol)), + ): + config = handler(config) # pyright: ignore + self.plugins = PluginRegistry(config.plugins) + + self._openapi_schema: OpenAPI | None = None + self._debug: bool = True + self.stores: StoreRegistry = ( + config.stores if isinstance(config.stores, StoreRegistry) else StoreRegistry(config.stores) + ) + self._lifespan_managers = config.lifespan + for store in self.stores._stores.values(): + self._lifespan_managers.append(store) + self._server_lifespan_managers = [p.server_lifespan for p in config.plugins or [] if isinstance(p, CLIPlugin)] + self.experimental_features = frozenset(config.experimental_features or []) + self.get_logger: GetLogger = get_logger_placeholder + self.logger: Logger | None = None + self.routes: list[HTTPRoute | ASGIRoute | WebSocketRoute] = [] + self.asgi_router = ASGIRouter(app=self) + + self.after_exception = [ensure_async_callable(h) for h in config.after_exception] + self.allowed_hosts = cast("AllowedHostsConfig | None", config.allowed_hosts) + self.before_send = [ensure_async_callable(h) for h in config.before_send] + self.compression_config = config.compression_config + self.cors_config = config.cors_config + self.csrf_config = config.csrf_config + self.event_emitter = config.event_emitter_backend(listeners=config.listeners) + self.logging_config = config.logging_config + self.multipart_form_part_limit = config.multipart_form_part_limit + self.on_shutdown = config.on_shutdown + self.on_startup = config.on_startup + self.openapi_config = config.openapi_config + self.request_class: type[Request] = config.request_class or Request + self.response_cache_config = config.response_cache_config + self.state = config.state + self._static_files_config = config.static_files_config + self.template_engine = config.template_config.engine_instance if config.template_config else None + self.websocket_class: type[WebSocket] = config.websocket_class or WebSocket + self.debug = config.debug + self.pdb_on_exception: bool = config.pdb_on_exception + self.include_in_schema = include_in_schema + + if self.pdb_on_exception: + warn_pdb_on_exception() + + try: + from starlette.exceptions import HTTPException as StarletteHTTPException + + from litestar.middleware.exceptions.middleware import _starlette_exception_handler + + config.exception_handlers.setdefault(StarletteHTTPException, _starlette_exception_handler) + except ImportError: + pass + + super().__init__( + after_request=config.after_request, + after_response=config.after_response, + before_request=config.before_request, + cache_control=config.cache_control, + dependencies=config.dependencies, + dto=config.dto, + etag=config.etag, + exception_handlers=config.exception_handlers, + guards=config.guards, + middleware=config.middleware, + opt=config.opt, + parameters=config.parameters, + path="", + request_class=self.request_class, + response_class=config.response_class, + response_cookies=config.response_cookies, + response_headers=config.response_headers, + return_dto=config.return_dto, + # route handlers are registered below + route_handlers=[], + security=config.security, + signature_namespace=config.signature_namespace, + signature_types=config.signature_types, + tags=config.tags, + type_encoders=config.type_encoders, + type_decoders=config.type_decoders, + include_in_schema=config.include_in_schema, + websocket_class=self.websocket_class, + ) + + for route_handler in config.route_handlers: + self.register(route_handler) + + if self.logging_config: + self.get_logger = self.logging_config.configure() + self.logger = self.get_logger("litestar") + + for static_config in self._static_files_config: + self.register(static_config.to_static_files_app()) + + self.asgi_handler = self._create_asgi_handler() + + @property + @deprecated(version="2.6.0", kind="property", info="Use create_static_files router instead") + def static_files_config(self) -> list[StaticFilesConfig]: + return self._static_files_config + + @property + @deprecated(version="2.0", alternative="Litestar.plugins.cli", kind="property") + def cli_plugins(self) -> list[CLIPluginProtocol]: + return list(self.plugins.cli) + + @property + @deprecated(version="2.0", alternative="Litestar.plugins.openapi", kind="property") + def openapi_schema_plugins(self) -> list[OpenAPISchemaPluginProtocol]: + return list(self.plugins.openapi) + + @property + @deprecated(version="2.0", alternative="Litestar.plugins.serialization", kind="property") + def serialization_plugins(self) -> list[SerializationPluginProtocol]: + return list(self.plugins.serialization) + + @staticmethod + def _get_default_plugins(plugins: list[PluginProtocol]) -> list[PluginProtocol]: + from litestar.plugins.core import MsgspecDIPlugin + + plugins.append(MsgspecDIPlugin()) + + with suppress(MissingDependencyException): + from litestar.contrib.pydantic import ( + PydanticDIPlugin, + PydanticInitPlugin, + PydanticPlugin, + PydanticSchemaPlugin, + ) + + pydantic_plugin_found = any(isinstance(plugin, PydanticPlugin) for plugin in plugins) + pydantic_init_plugin_found = any(isinstance(plugin, PydanticInitPlugin) for plugin in plugins) + pydantic_schema_plugin_found = any(isinstance(plugin, PydanticSchemaPlugin) for plugin in plugins) + pydantic_serialization_plugin_found = any(isinstance(plugin, PydanticDIPlugin) for plugin in plugins) + if not pydantic_plugin_found and not pydantic_init_plugin_found and not pydantic_schema_plugin_found: + plugins.append(PydanticPlugin()) + elif not pydantic_plugin_found and pydantic_init_plugin_found and not pydantic_schema_plugin_found: + plugins.append(PydanticSchemaPlugin()) + elif not pydantic_plugin_found and not pydantic_init_plugin_found: + plugins.append(PydanticInitPlugin()) + if not pydantic_plugin_found and not pydantic_serialization_plugin_found: + plugins.append(PydanticDIPlugin()) + with suppress(MissingDependencyException): + from litestar.contrib.attrs import AttrsSchemaPlugin + + pre_configured = any(isinstance(plugin, AttrsSchemaPlugin) for plugin in plugins) + if not pre_configured: + plugins.append(AttrsSchemaPlugin()) + return plugins + + @property + def debug(self) -> bool: + return self._debug + + @debug.setter + def debug(self, value: bool) -> None: + """Sets the debug logging level for the application. + + When possible, it calls the `self.logging_config.set_level` method. This allows for implementation specific code and APIs to be called. + """ + if self.logger and self.logging_config: + self.logging_config.set_level(self.logger, logging.DEBUG if value else logging.INFO) + elif self.logger and hasattr(self.logger, "setLevel"): # pragma: no cover + self.logger.setLevel(logging.DEBUG if value else logging.INFO) # pragma: no cover + if isinstance(self.logging_config, LoggingConfig): + self.logging_config.loggers["litestar"]["level"] = "DEBUG" if value else "INFO" + self._debug = value + + async def __call__( + self, + scope: Scope | LifeSpanScope, + receive: Receive | LifeSpanReceive, + send: Send | LifeSpanSend, + ) -> None: + """Application entry point. + + Lifespan events (startup / shutdown) are sent to the lifespan handler, otherwise the ASGI handler is used + + Args: + scope: The ASGI connection scope. + receive: The ASGI receive function. + send: The ASGI send function. + + Returns: + None + """ + if scope["type"] == "lifespan": + await self.asgi_router.lifespan(receive=receive, send=send) # type: ignore[arg-type] + return + + scope["app"] = self + scope.setdefault("state", {}) + await self.asgi_handler(scope, receive, self._wrap_send(send=send, scope=scope)) # type: ignore[arg-type] + + async def _call_lifespan_hook(self, hook: LifespanHook) -> None: + ret = hook(self) if inspect.signature(hook).parameters else hook() # type: ignore[call-arg] + + if is_async_callable(hook): # pyright: ignore[reportGeneralTypeIssues] + await ret + + @asynccontextmanager + async def lifespan(self) -> AsyncGenerator[None, None]: + """Context manager handling the ASGI lifespan. + + It will be entered when the ``lifespan`` message has been received from the + server, and exit after the ``asgi.shutdown`` message. During this period, it is + responsible for calling the ``on_startup``, ``on_shutdown`` hooks, as well as + custom lifespan managers. + """ + async with AsyncExitStack() as exit_stack: + for hook in self.on_shutdown[::-1]: + exit_stack.push_async_callback(partial(self._call_lifespan_hook, hook)) + + await exit_stack.enter_async_context(self.event_emitter) + + for manager in self._lifespan_managers: + if not isinstance(manager, AbstractAsyncContextManager): + manager = manager(self) + await exit_stack.enter_async_context(manager) + + for hook in self.on_startup: + await self._call_lifespan_hook(hook) + + yield + + @property + def openapi_schema(self) -> OpenAPI: + """Access the OpenAPI schema of the application. + + Returns: + The :class:`OpenAPI` + <pydantic_openapi_schema.open_api.OpenAPI> instance of the + application. + + Raises: + ImproperlyConfiguredException: If the application ``openapi_config`` attribute is ``None``. + """ + return self.plugins.get(OpenAPIPlugin).provide_openapi() + + @classmethod + def from_config(cls, config: AppConfig) -> Self: + """Initialize a ``Litestar`` application from a configuration instance. + + Args: + config: An instance of :class:`AppConfig` <.config.AppConfig> + + Returns: + An instance of ``Litestar`` application. + """ + return cls(**dict(extract_dataclass_items(config))) + + def register(self, value: ControllerRouterHandler) -> None: # type: ignore[override] + """Register a route handler on the app. + + This method can be used to dynamically add endpoints to an application. + + Args: + value: An instance of :class:`Router <.router.Router>`, a subclass of + :class:`Controller <.controller.Controller>` or any function decorated by the route handler decorators. + + Returns: + None + """ + routes = super().register(value=value) + + for route in routes: + route_handlers = get_route_handlers(route) + + for route_handler in route_handlers: + route_handler.on_registration(self) + + if isinstance(route, HTTPRoute): + route.create_handler_map() + + elif isinstance(route, WebSocketRoute): + route.handler_parameter_model = route.create_handler_kwargs_model(route.route_handler) + + for plugin in self.plugins.receive_route: + plugin.receive_route(route) + + self.asgi_router.construct_routing_trie() + + def get_handler_index_by_name(self, name: str) -> HandlerIndex | None: + """Receives a route handler name and returns an optional dictionary containing the route handler instance and + list of paths sorted lexically. + + Examples: + .. code-block:: python + + from litestar import Litestar, get + + + @get("/", name="my-handler") + def handler() -> None: + pass + + + app = Litestar(route_handlers=[handler]) + + handler_index = app.get_handler_index_by_name("my-handler") + + # { "paths": ["/"], "handler" ... } + + Args: + name: A route handler unique name. + + Returns: + A :class:`HandlerIndex <.app.HandlerIndex>` instance or ``None``. + """ + handler = self.asgi_router.route_handler_index.get(name) + if not handler: + return None + + identifier = handler.name or str(handler) + routes = self.asgi_router.route_mapping[identifier] + paths = sorted(unique([route.path for route in routes])) + + return HandlerIndex(handler=handler, paths=paths, identifier=identifier) + + def route_reverse(self, name: str, **path_parameters: Any) -> str: + """Receives a route handler name, path parameter values and returns url path to the handler with filled path + parameters. + + Examples: + .. code-block:: python + + from litestar import Litestar, get + + + @get("/group/{group_id:int}/user/{user_id:int}", name="get_membership_details") + def get_membership_details(group_id: int, user_id: int) -> None: + pass + + + app = Litestar(route_handlers=[get_membership_details]) + + path = app.route_reverse("get_membership_details", user_id=100, group_id=10) + + # /group/10/user/100 + + Args: + name: A route handler unique name. + **path_parameters: Actual values for path parameters in the route. + + Raises: + NoRouteMatchFoundException: If route with 'name' does not exist, path parameters are missing in + ``**path_parameters or have wrong type``. + + Returns: + A fully formatted url path. + """ + handler_index = self.get_handler_index_by_name(name) + if handler_index is None: + raise NoRouteMatchFoundException(f"Route {name} can not be found") + + allow_str_instead = {datetime, date, time, timedelta, float, Path} + routes = sorted( + self.asgi_router.route_mapping[handler_index["identifier"]], + key=lambda r: len(r.path_parameters), + reverse=True, + ) + passed_parameters = set(path_parameters.keys()) + + selected_route = next( + ( + route + for route in routes + if passed_parameters.issuperset({param.name for param in route.path_parameters}) + ), + routes[-1], + ) + output: list[str] = [] + for component in selected_route.path_components: + if isinstance(component, PathParameterDefinition): + val = path_parameters.get(component.name) + if not isinstance(val, component.type) and ( + component.type not in allow_str_instead or not isinstance(val, str) + ): + raise NoRouteMatchFoundException( + f"Received type for path parameter {component.name} doesn't match declared type {component.type}" + ) + output.append(str(val)) + else: + output.append(component) + + return join_paths(output) + + @deprecated( + "2.6.0", info="Use create_static_files router instead of StaticFilesConfig, which works with route_reverse" + ) + def url_for_static_asset(self, name: str, file_path: str) -> str: + """Receives a static files handler name, an asset file path and returns resolved url path to the asset. + + Examples: + .. code-block:: python + + from litestar import Litestar + from litestar.static_files.config import StaticFilesConfig + + app = Litestar( + static_files_config=[ + StaticFilesConfig(directories=["css"], path="/static/css", name="css") + ] + ) + + path = app.url_for_static_asset("css", "main.css") + + # /static/css/main.css + + Args: + name: A static handler unique name. + file_path: a string containing path to an asset. + + Raises: + NoRouteMatchFoundException: If static files handler with ``name`` does not exist. + + Returns: + A url path to the asset. + """ + + handler_index = self.get_handler_index_by_name(name) + if handler_index is None: + raise NoRouteMatchFoundException(f"Static handler {name} can not be found") + + handler_fn = cast("AnyCallable", handler_index["handler"].fn) + if not isinstance(handler_fn, StaticFiles): + raise NoRouteMatchFoundException(f"Handler with name {name} is not a static files handler") + + return join_paths([handler_index["paths"][0], file_path]) # type: ignore[unreachable] + + @property + def route_handler_method_view(self) -> dict[str, list[str]]: + """Map route handlers to paths. + + Returns: + A dictionary of router handlers and lists of paths as strings + """ + route_map: dict[str, list[str]] = { + handler: [route.path for route in routes] for handler, routes in self.asgi_router.route_mapping.items() + } + return route_map + + def _create_asgi_handler(self) -> ASGIApp: + """Create an ASGIApp that wraps the ASGI router inside an exception handler. + + If CORS or TrustedHost configs are provided to the constructor, they will wrap the router as well. + """ + asgi_handler: ASGIApp = self.asgi_router + if self.cors_config: + asgi_handler = CORSMiddleware(app=asgi_handler, config=self.cors_config) + + return wrap_in_exception_handler( + app=asgi_handler, + exception_handlers=self.exception_handlers or {}, # pyright: ignore + ) + + def _wrap_send(self, send: Send, scope: Scope) -> Send: + """Wrap the ASGI send and handles any 'before send' hooks. + + Args: + send: The ASGI send function. + scope: The ASGI scope. + + Returns: + An ASGI send function. + """ + if self.before_send: + + async def wrapped_send(message: Message) -> None: + for hook in self.before_send: + await hook(message, scope) + await send(message) + + return wrapped_send + return send + + def update_openapi_schema(self) -> None: + """Update the OpenAPI schema to reflect the route handlers registered on the app. + + Returns: + None + """ + self.plugins.get(OpenAPIPlugin)._build_openapi_schema() + + def emit(self, event_id: str, *args: Any, **kwargs: Any) -> None: + """Emit an event to all attached listeners. + + Args: + event_id: The ID of the event to emit, e.g ``my_event``. + args: args to pass to the listener(s). + kwargs: kwargs to pass to the listener(s) + + Returns: + None + """ + self.event_emitter.emit(event_id, *args, **kwargs) |