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:

A simple Hello World Litestar app#
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])
A simple Hello World Litestar app#
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.

        flowchart LR
    Startup[ASGI-Event: lifespan.startup] --> on_startup
    Shutdown[ASGI-Event: lifespan.shutdown] --> on_shutdown

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:

Startup and Shutdown#
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.

Handling a database connection#
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])
Handling a database connection#
from contextlib import asynccontextmanager
from collections.abc 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:

Example of multiple context managers and shutdown hooks#
app = Litestar(lifespan=[ctx_a, ctx_b], on_shutdown=[hook_a, hook_b])

During shutdown, they are executed in the following order:

        flowchart LR
    ctx_b --> ctx_a --> hook_a --> hook_b

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:

Using Application State#
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:

Using Application State#
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}))
Using Application State#
from typing import Any

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.

To do this, set deep_copy to True in the State constructor.

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:

Accessing application State in a handler function#
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:

Using Custom State to ensure immutability#
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])
Using Custom State to ensure immutability#
from typing import Any

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.

After Exception Hook#
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.

Before Send Hook#
from __future__ import annotations

from typing import TYPE_CHECKING

from litestar import Litestar, get
from litestar.datastructures import MutableScopeHeaders

if TYPE_CHECKING:

    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.

Example usage of the on_app_init hook to modify the application configuration.#
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:

  1. The application object

  2. Routers

  3. Controllers

  4. Handlers

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: