Litestar
Plugins¶
Litestar supports a plugin system that allows you to extend the functionality of the framework.
Plugins are defined by protocols, and any type that satisfies a protocol can be included in the plugins argument of
the app.
InitPlugin¶
InitPlugin defines an interface that allows for customization of the application’s initialization process.
Init plugins can define dependencies, add route handlers, configure middleware, and much more!
Implementations of these plugins must define a single method:
on_app_init(self, app_config: AppConfig) -> AppConfig:
The method accepts and must return an AppConfig instance, which can be modified
and is later used to instantiate the application.
This method is invoked after any on_app_init hooks have been called, and each plugin is invoked in the order that
they are provided in the plugins argument of the app. Because of this, plugin
authors should make it clear in their documentation if their plugin should be invoked before or after other plugins.
Example¶
The following example shows a simple plugin that adds a route handler, and a dependency to the application.
InitPlugin implementation example¶from litestar import Litestar, get
from litestar.config.app import AppConfig
from litestar.di import Provide
from litestar.plugins import InitPlugin
@get("/", sync_to_thread=False)
def route_handler(name: str) -> dict[str, str]:
return {"hello": name}
def get_name() -> str:
return "world"
class MyPlugin(InitPlugin):
def on_app_init(self, app_config: AppConfig) -> AppConfig:
app_config.dependencies["name"] = Provide(get_name, sync_to_thread=False)
app_config.route_handlers.append(route_handler)
return app_config
app = Litestar(plugins=[MyPlugin()])
Run it
> curl http://127.0.0.1:8000/
{"hello":"world"}
The MyPlugin class is an implementation of the InitPlugin. It
defines a single method, on_app_init(), which takes an AppConfig instance
as an argument and returns same.
In the on_app_init() method, the dependency mapping is updated to include a new dependency named "name", which
is provided by the get_name() function, and route_handlers is updated to include the route_handler()
function. The modified AppConfig instance is then returned.
SerializationPlugin¶
The SerializationPlugin defines a contract for plugins that
provide serialization functionality for data types that are otherwise unsupported by the
framework.
Implementations of these plugins must define the following methods.
The method takes a FieldDefinition instance as an argument and returns a bool
indicating whether the plugin supports serialization for that type.
This method accepts a FieldDefinition instance as an argument and must return a
AbstractDTO implementation that can be used to serialize and deserialize
the type.
During application startup, if a data or return annotation is encountered that is not a supported type, is supported by
the plugin, and doesn’t otherwise have a dto or return_dto defined, the plugin is used to create a DTO type for
that annotation.
Example¶
The following example shows a minimal SerializationPlugin implementation that mirrors the pattern used by
advanced-alchemy’s SQLAlchemySerializationPlugin:
SerializationPlugin implementation example¶from __future__ import annotations
from dataclasses import dataclass
from typing import get_args, get_origin
from litestar import Litestar, get
from litestar.dto import AbstractDTO, DataclassDTO
from litestar.plugins import SerializationPlugin
from litestar.typing import FieldDefinition
@dataclass
class Company:
id: int
name: str
class CompanySerializationPlugin(SerializationPlugin):
def __init__(self) -> None:
self._type_dto_map: dict[type, type[AbstractDTO]] = {}
@staticmethod
def _unwrap_collection(annotation: object) -> object:
if get_origin(annotation) in (list, tuple, set):
return next(iter(get_args(annotation)), annotation)
return annotation
def supports_type(self, field_definition: FieldDefinition) -> bool:
annotation = self._unwrap_collection(field_definition.annotation)
return isinstance(annotation, type) and issubclass(annotation, Company)
def create_dto_for_type(self, field_definition: FieldDefinition) -> type[AbstractDTO]:
annotation = self._unwrap_collection(field_definition.annotation)
assert isinstance(annotation, type)
if (cached := self._type_dto_map.get(annotation)) is not None:
return cached
dto_type: type[AbstractDTO] = DataclassDTO[annotation]
self._type_dto_map[annotation] = dto_type
return dto_type
@get("/", sync_to_thread=False)
def get_company() -> Company:
return Company(id=1, name="ACME")
app = Litestar(route_handlers=[get_company], plugins=[CompanySerializationPlugin()])
CompanySerializationPlugin.supports_type() returns a bool indicating whether the plugin should be used for
a given type. The example unwraps collection annotations (list[Company], tuple[Company, ...], etc.) so that a
handler returning either Company or a collection of Company instances is recognised as the same underlying model.
CompanySerializationPlugin.create_dto_for_type() receives the same FieldDefinition
and must return an AbstractDTO subclass. It first unwraps any collection
wrapper to recover the concrete model type, checks the _type_dto_map cache, and if no DTO has been built yet,
parametrises DataclassDTO with the annotation, caches it, and returns it. Caching keeps a single
DTO class per model so handlers that reference the same model type reuse the same serializer.
For a production-grade implementation using SQLAlchemy models instead of dataclasses, refer to the
advanced-alchemy library documentation.
DIPlugin¶
DIPlugin can be used to extend Litestar’s dependency
injection by providing information about injectable types.
Its main purpose it to facilitate the injection of callables with unknown signatures,
for example Pydantic’s BaseModel classes; These are not supported natively since,
while they are callables, their type information is not contained within their callable
signature (their __init__() method).
from inspect import Parameter, Signature
from typing import Any
from litestar import Litestar, get
from litestar.di import Provide
from litestar.plugins import DIPlugin
class MyBaseType:
def __init__(self, param):
self.param = param
class MyDIPlugin(DIPlugin):
def has_typed_init(self, type_: Any) -> bool:
return issubclass(type_, MyBaseType)
def get_typed_init(self, type_: Any) -> tuple[Signature, dict[str, Any]]:
signature = Signature([Parameter(name="param", kind=Parameter.POSITIONAL_OR_KEYWORD)])
annotations = {"param": str}
return signature, annotations
@get("/", dependencies={"injected": Provide(MyBaseType, sync_to_thread=False)})
async def handler(injected: MyBaseType) -> str:
return injected.param
app = Litestar(route_handlers=[handler], plugins=[MyDIPlugin()])
Run it
> curl 'http://127.0.0.1:8000/?param=hello'
hello
ReceiveRoutePlugin¶
ReceiveRoutePlugin allows you to receive routes as they are registered on the application.
This can be useful for plugins that need to perform actions based on the routes being added, such as generating
documentation, validating route configurations, or tracking route statistics.
Implementations of this plugin must define a single method:
receive_route(self, route: BaseRoute) -> None:
The method receives a BaseRoute instance as routes are registered on the application.
This happens during the application initialization process, after routes are created but before the application starts.
Example¶
The following example shows a simple plugin that logs information about each route as it’s registered:
from litestar.plugins import ReceiveRoutePlugin
from litestar.routes import BaseRoute
class RouteLoggerPlugin(ReceiveRoutePlugin):
def receive_route(self, route: BaseRoute) -> None:
print(f"Route registered: {route.path} [{', '.join(route.methods)}]")