Applications#
Application objects#
At the root of every Litestar application is an instance of the Litestar
class. Typically, this code will be placed in a file called main.py
, app.py
, asgi.py
or similar
at the project’s root directory.
These entry points are also used during CLI autodiscovery
Creating an app is straightforward – the only required args is a list
of Controllers
, Routers
,
or Route handlers
:
from typing import Dict
from litestar import Litestar, get
@get("/")
async def hello_world() -> Dict[str, str]:
"""Handler function that returns a greeting dictionary."""
return {"hello": "world"}
app = Litestar(route_handlers=[hello_world])
Run it
> curl http://127.0.0.1:8000/
{"hello":"world"}
The app instance 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.
See also
To learn more about registering routes, check out this chapter in the documentation:
Startup and Shutdown#
You can pass a list of callables - either sync or async functions, methods, or class instances
- to the on_startup
/ on_shutdown
kwargs of the app
instance. Those will be called in
order, once the ASGI server such as uvicorn,
Hypercorn, Granian,
Daphne, etc. emits the respective event.
A classic use case for this is database connectivity. Often, we want to establish a database connection on application startup, and then close it gracefully upon shutdown.
For example, let us create a database connection using the async engine from
SQLAlchemy. We create two functions, one to get or
establish the connection, and another to close it, and then pass them to the Litestar
constructor:
import os
from typing import cast
from sqlalchemy.ext.asyncio import AsyncEngine, create_async_engine
from litestar import Litestar
DB_URI = os.environ.get("DATABASE_URI", "postgresql+asyncpg://postgres:mysecretpassword@pg.db:5432/db")
def get_db_connection(app: Litestar) -> AsyncEngine:
"""Returns the db engine.
If it doesn't exist, creates it and saves it in on the application state object
"""
if not getattr(app.state, "engine", None):
app.state.engine = create_async_engine(DB_URI)
return cast("AsyncEngine", app.state.engine)
async def close_db_connection(app: Litestar) -> None:
"""Closes the db connection stored in the application State object."""
if getattr(app.state, "engine", None):
await cast("AsyncEngine", app.state.engine).dispose()
app = Litestar(on_startup=[get_db_connection], on_shutdown=[close_db_connection])
Lifespan context managers#
In addition to the lifespan hooks, Litestar also supports managing the lifespan of an application using an asynchronous context manager. This can be useful when dealing with long running tasks, or those that need to keep a certain context object, such as a connection, around.
from contextlib import asynccontextmanager
from typing import AsyncGenerator
from sqlalchemy.ext.asyncio import create_async_engine
from litestar import Litestar
@asynccontextmanager
async def db_connection(app: Litestar) -> AsyncGenerator[None, None]:
engine = getattr(app.state, "engine", None)
if engine is None:
engine = create_async_engine("postgresql+asyncpg://postgres:mysecretpassword@pg.db:5432/db")
app.state.engine = engine
try:
yield
finally:
await engine.dispose()
app = Litestar(lifespan=[db_connection])
Order of execution#
When multiple lifespan context managers and on_shutdown
hooks are specified,
Litestar will invoke the context managers in inverse order before the
shutdown hooks are invoked.
Consider the case where there are two lifespan context managers ctx_a
and ctx_b
as well as two shutdown hooks
hook_a
and hook_b
as shown in the following code:
app = Litestar(lifespan=[ctx_a, ctx_b], on_shutdown=[hook_a, hook_b])
During shutdown, they are executed in the following order:
As seen, the context managers are invoked in inverse order. On the other hand, the shutdown hooks are invoked in their specified order.
Using Application State#
As seen in the examples for the on_startup / on_shutdown, callables
passed to these hooks can receive an optional kwarg called app
, through which the application’s
state object and other properties can be accessed. The advantage of using application state
,
is that it can be accessed during multiple stages of the connection, and it can be injected into dependencies and
route handlers.
The Application State is an instance of the datastructures.state.State
datastructure, and it is
accessible via the state
attribute. As such it can be accessed wherever the app instance
is accessible.
state
is one of the
reserved keyword arguments.
It is important to understand in this context that the application instance is injected into the ASGI scope
mapping
for each connection (i.e. request or websocket connection) as scope["app"]
. This makes the application
accessible wherever the scope mapping is available, e.g. in middleware, on Request
and
WebSocket
instances (accessible as request.app
/ socket.app
), and many
other places.
Therefore, state
offers an easy way to share contextual data between disparate parts
of the application, as seen below:
import logging
from typing import TYPE_CHECKING, Any
from litestar import Litestar, Request, get
from litestar.datastructures import State
from litestar.di import Provide
if TYPE_CHECKING:
from litestar.types import ASGIApp, Receive, Scope, Send
logger = logging.getLogger(__name__)
def set_state_on_startup(app: Litestar) -> None:
"""Startup and shutdown hooks can receive `State` as a keyword arg."""
app.state.value = "abc123"
def middleware_factory(*, app: "ASGIApp") -> "ASGIApp":
"""A middleware can access application state via `scope`."""
async def my_middleware(scope: "Scope", receive: "Receive", send: "Send") -> None:
state = scope["app"].state
logger.info("state value in middleware: %s", state.value)
await app(scope, receive, send)
return my_middleware
async def my_dependency(state: State) -> Any:
"""Dependencies can receive state via injection."""
logger.info("state value in dependency: %s", state.value)
@get("/", dependencies={"dep": Provide(my_dependency)}, middleware=[middleware_factory], sync_to_thread=False)
def get_handler(state: State, request: Request, dep: Any) -> None:
"""Handlers can receive state via injection."""
logger.info("state value in handler from `State`: %s", state.value)
logger.info("state value in handler from `Request`: %s", request.app.state.value)
app = Litestar(route_handlers=[get_handler], on_startup=[set_state_on_startup])
Initializing Application State#
To seed application state, you can pass a State
object to the
state
parameter of the Litestar constructor:
from typing import Any, Dict
from litestar import Litestar, get
from litestar.datastructures import State
@get("/", sync_to_thread=False)
def handler(state: State) -> Dict[str, Any]:
return state.dict()
app = Litestar(route_handlers=[handler], state=State({"count": 100}))
Note
State
can be initialized with a dictionary
, an instance of
ImmutableState
or State
,
or a list
of tuples
containing key/value pairs.
You may instruct State
to deep copy initialized data to prevent mutation from outside
the application context.
Injecting Application State into Route Handlers and Dependencies#
As seen in the above example, Litestar offers an easy way to inject state into route handlers and dependencies - simply
by specifying state
as a kwarg to the handler or dependency function. For example:
from litestar import get
from litestar.datastructures import State
@get("/")
def handler(state: State) -> None: ...
When using this pattern you can specify the class to use for the state object. This type is not merely for type
checkers, rather Litestar will instantiate a new state
instance based on the type you set there.
This allows users to use custom classes for State
.
While this is very powerful, it might encourage users to follow anti-patterns: it is important to emphasize that using
state can lead to code that is hard to reason about and bugs that are difficult to understand, due to changes in
different ASGI contexts. As such, this pattern should be used only when it is the best choice and in a limited fashion.
To discourage its use, Litestar also offers a builtin ImmutableState
class.
You can use this class to type state and ensure that no mutation of state is allowed:
from typing import Any, Dict
from litestar import Litestar, get
from litestar.datastructures import ImmutableState
@get("/", sync_to_thread=False)
def handler(state: ImmutableState) -> Dict[str, Any]:
setattr(state, "count", 1) # raises AttributeError
return state.dict()
app = Litestar(route_handlers=[handler])
Application Hooks#
Litestar includes several application level hooks that allow users to run their own sync or async callables. While you are free to use these hooks as you see fit, the design intention behind them is to allow for easy instrumentation for observability (monitoring, tracing, logging, etc.).
Note
All application hook kwargs detailed below receive either a single callable or a list
of callables.
If a list
is provided, it is called in the order it is given.
After Exception#
The after_exception
hook takes a
sync or async callable
that is called with two arguments:
the exception
that occurred and the ASGI scope
of the request or websocket connection.
import logging
from typing import TYPE_CHECKING
from litestar import Litestar, get
from litestar.exceptions import HTTPException
from litestar.status_codes import HTTP_400_BAD_REQUEST
logger = logging.getLogger()
if TYPE_CHECKING:
from litestar.types import Scope
@get("/some-path", sync_to_thread=False)
def my_handler() -> None:
"""Route handler that raises an exception."""
raise HTTPException(detail="bad request", status_code=HTTP_400_BAD_REQUEST)
async def after_exception_handler(exc: Exception, scope: "Scope") -> None:
"""Hook function that will be invoked after each exception."""
state = scope["app"].state
if not hasattr(state, "error_count"):
state.error_count = 1
else:
state.error_count += 1
logger.info(
"an exception of type %s has occurred for requested path %s and the application error count is %d.",
type(exc).__name__,
scope["path"],
state.error_count,
)
app = Litestar([my_handler], after_exception=[after_exception_handler])
Attention
This hook is not meant to handle exceptions - it just receives them to allow for side effects. To handle exceptions you should define exception handlers.
Before Send#
The before_send
hook takes a
sync or async callable
that is called when an ASGI message is
sent. The hook receives the message instance and the ASGI scope
.
from __future__ import annotations
from typing import TYPE_CHECKING
from litestar import Litestar, get
from litestar.datastructures import MutableScopeHeaders
if TYPE_CHECKING:
from typing import Dict
from litestar.types import Message, Scope
@get("/test", sync_to_thread=False)
def handler() -> Dict[str, str]:
"""Example Handler function."""
return {"key": "value"}
async def before_send_hook_handler(message: Message, scope: Scope) -> None:
"""The function will be called on each ASGI message.
We therefore ensure it runs only on the message start event.
"""
if message["type"] == "http.response.start":
headers = MutableScopeHeaders.from_message(message=message)
headers["My Header"] = scope["app"].state.message
def on_startup(app: Litestar) -> None:
"""A function that will populate the app state before any requests are received."""
app.state.message = "value injected during send"
app = Litestar(route_handlers=[handler], on_startup=[on_startup], before_send=[before_send_hook_handler])
Initialization#
Litestar includes a hook for intercepting the arguments passed to the
Litestar constructor
, before they are used to instantiate the application.
Handlers can be passed to the on_app_init
parameter on construction of the application,
and in turn, each will receive an instance of AppConfig
and must return an instance of same.
This hook is useful for applying common configuration between applications, and for use by developers who may wish to develop third-party application configuration systems.
Note
on_app_init
handlers cannot be Coroutine function definition functions, as they are
called within __init__
, outside of an async context.
from typing import TYPE_CHECKING
from litestar import Litestar
if TYPE_CHECKING:
from litestar.config.app import AppConfig
async def close_db_connection() -> None:
"""Closes the database connection on application shutdown."""
def receive_app_config(app_config: "AppConfig") -> "AppConfig":
"""Receives parameters from the application.
In reality, this would be a library of boilerplate that is carried from one application to another, or a third-party
developed application configuration tool.
"""
app_config.on_shutdown.append(close_db_connection)
return app_config
app = Litestar([], on_app_init=[receive_app_config])
Layered architecture#
Litestar has a layered architecture compromising of 4 layers:
There are many parameters that can be defined on every layer, in which case the parameter defined on the layer closest to the handler takes precedence. This allows for maximum flexibility and simplicity when configuring complex applications and enables transparent overriding of parameters.
Parameters that support layering are: