FastAPI#

Layered configuration#

Litestar uses a layered architecture. The parts used to group routes can also be used to hierarchically organize an application. Settings and configuration like dependencies, exception handlers, guards, middleware, response cookies and headers, lifecycle hooks, OpenAPI configuration and many more can be defined on any layer, and will be merged upon registration.

The layers in hierarchical order are

  1. Application (Litestar)

  2. Router / Controller (these are on the same level and can be arbitrarily nested)

  3. Handler (BaseRouteHandler)

When the same configuration is set on multiple layers, the value set closest to the handler takes precedence.

Pydantic support#

Litestar does support Pydantic, but other than FastAPI, it is mostly agnostic about the modelling library you use. Litestar internally uses msgspec, but also ships with support for Pydantic, attrs and all builtin container types such as dataclasses or TypedDicts.

These, as well as msgspec, are supported through its plugin system ( SerializationPlugin and OpenAPISchemaPlugin) so it’s easy to add first-class support for any library.

Route handlers#

Litestar does not expose decorators on the application or router. A route is declared by a route handler (function with one of the method decorators, or a method on a Controller) and then registered on a Litestar or Router instance.

from fastapi import FastAPI

app = FastAPI()

@app.get("/")
async def index() -> dict[str, str]: ...
from starlette.applications import Starlette
from starlette.routing import Route

async def index(request): ...

routes = [Route("/", endpoint=index)]
app = Starlette(routes=routes)
from litestar import Litestar, get


@get("/")
async def index() -> dict[str, str]:
    return {"hello": "world"}


app = Litestar(route_handlers=[index])

Routers#

A few small differences between Litestar’s Router and FastAPI / Starlette’s:

  • A Litestar Router is not itself an ASGI application. It groups handlers and options. Routers are “flattened” into simple handlers during registration

  • A Litestar Router does not expose decorators

  • A Litestar Router does not run lifecycle hooks. Lifecycle hooks can only be registered on the application. See Life Cycle Hooks for more information

Host-based routing#

Litestar does not support dispatching requests by the Host header. Run each host as its own application behind a reverse proxy such as nginx or traefik.

Lifespan#

Litestar uses the async context manager pattern as Starlette/FastAPI:

from contextlib import asynccontextmanager

@asynccontextmanager
async def lifespan(app: FastAPI):
    # setup
    yield
    # teardown

app = FastAPI(lifespan=lifespan)
from collections.abc import AsyncIterator
from contextlib import asynccontextmanager

from litestar import Litestar, get


@asynccontextmanager
async def lifespan(app: Litestar) -> AsyncIterator[None]:
    # Setup code runs before the application starts accepting requests.
    yield
    # Teardown code runs after the application has stopped.


@get("/")
async def index() -> dict[str, str]:
    return {"hello": "world"}


app = Litestar(route_handlers=[index], lifespan=[lifespan])

Application state#

Application-scoped data lives on State, the equivalent of FastAPI’s app.state. State seeded on the application is available to dependencies through the state parameter and to handlers through request.app.state. Per-request data lives on request.state instead, and is wiped between requests.

from fastapi import Request

async def get_arq_redis(request: Request) -> ArqRedis:
    return request.state.arq_redis
from litestar import Litestar, get
from litestar.datastructures import State
from litestar.di import Provide


class ArqRedis:
    """Stand-in for an ``arq`` Redis client."""


async def get_arq_redis(state: State) -> ArqRedis:
    return state.arq_redis


@get("/", dependencies={"arq_redis": Provide(get_arq_redis)})
async def handler(arq_redis: ArqRedis) -> dict[str, str]:
    return {"type": type(arq_redis).__name__}


app = Litestar(
    route_handlers=[handler],
    state=State({"arq_redis": ArqRedis()}),
)

Handlers and request data#

Path and query parameters#

FastAPI allows declaring path and query parameters implicitly; If a function parameter of the same name is detected, the path / query parameter will be injected into it. Litestar uses the same mechanism but makes it explicit: The corresponding parameters must be marked with FromPath and FromQuery respectively:

@app.get("/{some_path}")
async def handler(some_path: str, some_query: int) -> None:
    return None
from litestar import get
from litestar.params import FromPath, FromQuery


@get("/{some_path:str}")
async def handler(some_path: FromPath[str], some_query: FromQuery[int]) -> None:
    return None

To define constraints or extend the OpenAPI schema for parameters, use their Annotated shapes PathParameter and QueryParameter:

@app.get("/{some_path}")
async def handler(
    some_path: str,
    some_query: Annotated[int, Query(gt=1)],
) -> None:
    return None
from typing import Annotated

from litestar import get
from litestar.params import CookieParameter, HeaderParameter


@get("/")
async def handler(
    some_cookie: Annotated[int, CookieParameter(lt=10)],
    some_header: Annotated[int, HeaderParameter(name="some-header", gt=1)],
) -> None:
    return None

See Parameters for more details.

JSON request body#

The JSON body binds is injected into the handler via the data parameter. Its type can be any type supported by Litestar or a validation extension. The data will be validated against this type before it is passed to the handler.

Litestar natively supports builtin types, dataclasses, Pydantic models, msgspec Structs and attrs classes.

class Item(BaseModel):
    name: str

@app.post("/items/")
async def create_item(item: Item) -> dict[str, str]:
    return {"name": item.name}
from pydantic import BaseModel

from litestar import Litestar, post


class Item(BaseModel):
    name: str


@post("/items/")
async def create_item(data: Item) -> dict[str, str]:
    return {"name": data.name}


app = Litestar(route_handlers=[create_item])

Form data#

Form-encoded bodies use the same data parameter, annotated with Body() and a media_type of RequestEncodingType.URL_ENCODED or RequestEncodingType.MULTI_PART

One data declaration replaces FastAPI’s per-field Form() calls.

from fastapi import Form

@app.post("/login")
async def login(
    username: str = Form(...),
    password: str = Form(...),
) -> dict[str, str]:
    return {"user": username}
from typing import Annotated

from litestar import Litestar, post
from litestar.enums import RequestEncodingType
from litestar.params import Body


@post("/login")
async def login(
    data: Annotated[dict[str, str], Body(media_type=RequestEncodingType.URL_ENCODED)],
) -> dict[str, str]:
    return {"user": data["username"]}


app = Litestar(route_handlers=[login])

File uploads#

Multipart bodies use the same data parameter with media_type set to RequestEncodingType.MULTI_PART. Uploaded files arrive as UploadFile instances. To receive multiple files, you can simply use list[UploadFile]. For mixed form data (i.e. plain fields + uploads), you can simply annotate any field of a supported model type with UploadData.

from fastapi import UploadFile

@app.post("/upload/")
async def upload(files: list[UploadFile]) -> dict[str, list[str]]:
    return {"file_names": [f.filename for f in files]}
from typing import Annotated

from litestar import Litestar, post
from litestar.datastructures import UploadFile
from litestar.enums import RequestEncodingType
from litestar.params import Body


@post("/upload/")
async def upload_file(
    data: Annotated[list[UploadFile], Body(media_type=RequestEncodingType.MULTI_PART)],
) -> dict[str, list[str | None]]:
    return {"file_names": [file.filename for file in data]}


app = Litestar(route_handlers=[upload_file])

Synchronous handlers#

FastAPI inspects the handler at registration and runs synchronous callables on a threadpool. Litestar has the same mechanism, but it must be explicitly enabled by setting sync_to_thread=True. A synchronous handler without sync_to_thread will emit a warning at startup. If you intentionally do not want to run a synchronous handler inside a thread pool, set sync_to_thread=False.

@app.get("/")
def slow_handler() -> dict[str, str]:
    # implicitly run in a threadpool
    return {"hello": "world"}
import time

from litestar import Litestar, get


@get("/", sync_to_thread=True)
def slow_handler() -> dict[str, str]:
    time.sleep(0.01)
    return {"hello": "world"}


@get("/fast", sync_to_thread=False)
def fast_handler() -> dict[str, str]:
    return {"hello": "world"}


app = Litestar(route_handlers=[slow_handler, fast_handler])

Producing responses#

Default status codes#

FastAPI returns 200 for every method by default. Litestar picks the status code from the HTTP method: POST defaults to 201 Created, DELETE to 204 No Content, and everything else to 200. You can pass status_code= to the handler to override the default status code.

Cookies and headers#

FastAPI defaults to setting headers directly on a Response . Litestar offers two paths: declare static values on the decorator with response_cookies and response_headers, or return a Response with cookies= and headers= when the values depend on the request.

from fastapi import Response

@app.get("/")
async def index(response: Response) -> dict[str, str]:
    response.set_cookie(key="my-cookie", value="cookie-value")
    return {}
from litestar import Litestar, get
from litestar.datastructures import Cookie


@get("/", response_cookies=[Cookie(key="my-cookie", value="cookie-value")])
async def handler() -> str:
    return "hello"


app = Litestar(route_handlers=[handler])

Serialization#

The handler’s return annotation drives serialization. Structured types serialise to JSON; str returns text/plain; a typed response (Stream, Template, Redirect, File) picks its own media type. media_type= on the decorator overrides the default.

Templates#

FastAPI renders templates through Starlette’s Jinja2Templates helper and returns a TemplateResponse. In Litestar the engine is configured once on the application through TemplateConfig, and each handler returns a Template.

from fastapi.templating import Jinja2Templates

templates = Jinja2Templates(directory="templates")

@app.get("/uploads")
async def get_uploads(request: Request):
    return templates.TemplateResponse(
        "uploads.html", {"request": request, "debug": True}
    )
from pathlib import Path

from litestar import Litestar, get
from litestar.plugins.jinja import JinjaTemplateEngine
from litestar.response import Template
from litestar.template.config import TemplateConfig


@get("/uploads")
async def get_uploads() -> Template:
    return Template(
        template_str="<p>debug={{ debug }}</p>",
        context={"debug": True},
    )


app = Litestar(
    route_handlers=[get_uploads],
    template_config=TemplateConfig(
        directory=Path(__file__).parent,
        engine=JinjaTemplateEngine,
    ),
)

See also

Streaming responses#

FastAPI uses StreamingResponse; Litestar uses Stream. Both wrap a sync or async iterator.

from fastapi.responses import StreamingResponse

@app.get("/numbers")
async def stream_numbers() -> StreamingResponse:
    async def numbers():
        for i in range(5):
            yield f"{i}\n".encode()

    return StreamingResponse(numbers())
from collections.abc import AsyncIterator

from litestar import Litestar, get
from litestar.response import Stream


async def numbers() -> AsyncIterator[bytes]:
    for i in range(5):
        yield f"{i}\n".encode()


@get("/numbers")
async def stream_numbers() -> Stream:
    return Stream(numbers())


app = Litestar(route_handlers=[stream_numbers])

Dependency injection#

FastAPI declares a dependency at the call site, either as a default value or inside the parameter annotation via Depends(fn), or in a list on the path operation. Litestar declares dependencies as a mapping from parameter name to provider, attached to the dependencies keyword on any layer. Async callables can be used as dependency providers natively, sync callables must be wrapped in Provide.

from fastapi import FastAPI, Depends, APIRouter

async def route_dependency() -> bool: ...
async def nested_dependency() -> str: ...
async def router_dependency() -> int: ...
async def app_dependency(data: str = Depends(nested_dependency)) -> int: ...

router = APIRouter(dependencies=[Depends(router_dependency)])
app = FastAPI(dependencies=[Depends(nested_dependency)])
app.include_router(router)

@app.get("/")
async def handler(
    val_route: bool = Depends(route_dependency),
    val_router: int = Depends(router_dependency),
    val_nested: str = Depends(nested_dependency),
    val_app: int = Depends(app_dependency),
) -> None: ...
from litestar import Litestar, Router, get


async def route_dependency() -> bool:
    return True


async def nested_dependency() -> str:
    return "nested"


async def router_dependency() -> int:
    return 1


async def app_dependency(val_nested: str) -> int:
    return len(val_nested)


@get("/", dependencies={"val_route": route_dependency})
async def handler(val_route: bool, val_router: int, val_nested: str, val_app: int) -> dict[str, object]:
    return {
        "val_route": val_route,
        "val_router": val_router,
        "val_nested": val_nested,
        "val_app": val_app,
    }


router = Router(
    path="/",
    route_handlers=[handler],
    dependencies={"val_router": router_dependency},
)

app = Litestar(
    route_handlers=[router],
    dependencies={
        "val_app": app_dependency,
        "val_nested": nested_dependency,
    },
)

Note

As most configuration in Litestar, dependencies are layered. Inner layers override outer ones, so a router-scoped dependency replaces an application-scoped one of the same name, and a handler-scoped one replaces both.

Dependency overrides in tests#

FastAPI exposes app.dependency_overrides for swapping dependencies on a live application. Litestar does not. The omission is deliberate: an override that mutates a hared application is hard to reason about under parallel tests, and Litestar is of the opinion that it’s better to structure the application in such a way that overrides are not necessary.

Alternative strategies are, order of preference:

  1. Build the application from a factory that takes the dependencies your tests need to swap. Natural choice when the override exists to replace a real database with a test one

  2. Construct a fresh application per test with create_test_client(), passing the overridden dependency through the same dependencies= keyword the production application uses

  3. Patch the dependency callable with a mocking library: A last resort reserved for cases where first two are impractical reach

Exceptions and error responses#

Both frameworks export an HTTPException. The FastAPI one takes the status code as a positional argument; the Litestar one uses the keyword argument status_code. A positional argument to Litestar’s HTTPException is appended to detail, which is rendered in the JSON response.

from fastapi import HTTPException

@app.get("/")
async def index() -> None:
    raise HTTPException(400, detail="can't find that")
from litestar import Litestar, get
from litestar.exceptions import HTTPException


@get("/")
async def index() -> None:
    response_fields = {"array": "value"}
    raise HTTPException(
        status_code=400,
        detail=f"can't get that field: {response_fields.get('missing')}",
    )


app = Litestar(route_handlers=[index])

Custom exception handlers#

FastAPI registers exception handlers through @app.exception_handler. Litestar accepts a mapping from exception class or status code to a handler callable.

Tip

Exception handling is part of Litestar’s layered architecture, so these can be declared on applications, routers, controller or route handlers

class ItemNotFound(Exception): ...

@app.exception_handler(ItemNotFound)
async def handle(request, exc) -> JSONResponse:
    return JSONResponse({"detail": "item not found"}, status_code=404)
from litestar import Litestar, Request, Response, get


class ItemNotFoundError(Exception):
    pass


def handle_item_not_found(request: Request, exception: ItemNotFoundError) -> Response[dict[str, str]]:
    return Response({"detail": "item not found"}, status_code=404)


@get("/")
async def index() -> None:
    raise ItemNotFoundError


app = Litestar(
    route_handlers=[index],
    exception_handlers={ItemNotFoundError: handle_item_not_found},
)

Authentication#

FastAPI usually handles authentication through dependency injection. The same approach work in Litestar, but the more idiomatic choice is a guard or a custom Implementing Custom Authentication.

Info

A guard is a callable that receives the ASGIConnection and the BaseRouteHandler; raising an exception here aborts the request.

Tip

Guards are part of Litestar’s layered architecture, so these can be declared on applications, routers, controller or route handlers

from fastapi import Depends

async def authenticate(request: Request) -> None: ...

@app.get("/", dependencies=[Depends(authenticate)])
async def index() -> dict[str, str]: ...
from litestar import Litestar, get
from litestar.connection import ASGIConnection
from litestar.exceptions import NotAuthorizedException
from litestar.handlers import BaseRouteHandler


async def require_token(connection: ASGIConnection, route_handler: BaseRouteHandler) -> None:
    if connection.headers.get("authorization") != "Bearer secret":
        raise NotAuthorizedException


@get("/", guards=[require_token])
async def index() -> dict[str, str]:
    return {"hello": "world"}


app = Litestar(route_handlers=[index])

See also

Middleware#

Pure ASGI middleware - a callable that wraps an ASGI app and returns another - works the same in both frameworks. Middleware built on Starlette’s BaseHTTPMiddleware does not port directly: use ASGIMiddleware instead. Subclass it and implement handle(), which receives scope, receive, send, and a next_app callable (the equivalent of call_next).

from starlette.middleware.base import BaseHTTPMiddleware

class ProcessTimeMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next):
        start = time.monotonic()
        response = await call_next(request)
        response.headers["x-process-time"] = f"{time.monotonic() - start:.4f}"
        return response

app.add_middleware(ProcessTimeMiddleware)
import time

from litestar import Litestar, get
from litestar.datastructures import MutableScopeHeaders
from litestar.middleware import ASGIMiddleware
from litestar.types import ASGIApp, Message, Receive, Scope, Send


class ProcessTimeMiddleware(ASGIMiddleware):
    async def handle(self, scope: Scope, receive: Receive, send: Send, next_app: ASGIApp) -> None:
        start_time = time.monotonic()

        async def send_wrapper(message: Message) -> None:
            if message["type"] == "http.response.start":
                headers = MutableScopeHeaders.from_message(message=message)
                headers["x-process-time"] = f"{time.monotonic() - start_time:.4f}"
            await send(message)

        await next_app(scope, receive, send_wrapper)


@get("/")
async def index() -> dict[str, str]:
    return {"hello": "world"}


app = Litestar(route_handlers=[index], middleware=[ProcessTimeMiddleware()])

Background tasks#

FastAPI injects a BackgroundTasks object into the handler. In Litestar, a BackgroundTask (or a BackgroundTasks collection) can be passed directly to the response. These tasks run after the response body has been sent.

from fastapi import BackgroundTasks

@app.get("/")
async def greeter(tasks: BackgroundTasks) -> dict[str, str]:
    tasks.add_task(log_visit, "world")
    return {"hello": "world"}
import logging

from litestar import Litestar, Response, get
from litestar.background_tasks import BackgroundTask

logger = logging.getLogger(__name__)


async def log_visit(name: str) -> None:
    logger.info("greeted %s", name)


@get("/")
async def greeter() -> Response[dict[str, str]]:
    return Response(
        {"hello": "world"},
        background=BackgroundTask(log_visit, "world"),
    )


app = Litestar(route_handlers=[greeter])

WebSockets#

FastAPI exposes a single @app.websocket decorator for direct connection handling. Litestar offers three handler styles for the three patterns that recur in WebSocket code:

  • websocket(): Raw ASGI WebSocket handling, same semantics as @app.websocket

  • websocket_listener(): Per-message callback style that takes and returns typed values; the framework handles accept, the receive loop, and serialisation.

  • websocket_stream(): Async generator to produce messages that are pushed to the WebSocket; the framework handles accept, the receive loop, and serialisation.

from fastapi import WebSocket

@app.websocket("/ws")
async def echo(socket: WebSocket) -> None:
    await socket.accept()
    while True:
        data = await socket.receive_json()
        await socket.send_json({"echo": data["message"]})
from litestar import Litestar, websocket_listener


@websocket_listener("/ws")
async def echo(data: dict[str, str]) -> dict[str, str]:
    return {"echo": data["message"]}


app = Litestar(route_handlers=[echo])

For broadcast and pub/sub patterns, pair these handlers with the channels plugin. Channels handles per-channel subscriptions, history, and inter-process fan-out through a pluggable broker, and can generate WebSocket route handlers that publish incoming events to subscribed clients.

See also

Testing#

Both frameworks ship an httpx-based TestClient. The Litestar one is TestClient. For unit testing, Litestar also provides create_test_client() and create_async_test_client(), which take the same arguments as Litestar and return a configured client.

from fastapi.testclient import TestClient

def test_index() -> None:
    with TestClient(app) as client:
        response = client.get("/")
    assert response.status_code == 200
from litestar import get
from litestar.status_codes import HTTP_200_OK
from litestar.testing import create_test_client


@get("/", sync_to_thread=False)
def hello() -> dict[str, str]:
    return {"hello": "world"}


def test_hello() -> None:
    with create_test_client(route_handlers=[hello]) as client:
        response = client.get("/")
    assert response.status_code == HTTP_200_OK
    assert response.json() == {"hello": "world"}

See also

OpenAPI customisation#

Per-route OpenAPI metadata (tags, summary, description,``operation_id``, etc.) sits on the handler decorator in both frameworks. Application-wide options live on OpenAPIConfig, passed as openapi_config= to the application.

@app.get(
    "/items/{item_id}",
    tags=["items"],
    summary="Retrieve an item",
    description="Look up a single item by its numeric identifier.",
    operation_id="get_item_by_id",
)
async def get_item(item_id: int) -> dict[str, int]:
    return {"id": item_id}
from litestar import Litestar, get
from litestar.openapi.config import OpenAPIConfig
from litestar.params import FromPath


@get(
    "/items/{item_id:int}",
    tags=["items"],
    summary="Retrieve an item",
    description="Look up a single item by its numeric identifier.",
    operation_id="get_item_by_id",
)
async def get_item(item_id: FromPath[int]) -> dict[str, int]:
    return {"id": item_id}


app = Litestar(
    route_handlers=[get_item],
    openapi_config=OpenAPIConfig(title="Items API", version="1.0.0"),
)

See also

Quick reference#

A lookup table for the most common translations.

Concept

FastAPI / Starlette

Litestar

Route declaration

@app.get("/")

@get("/") + Litestar([handler])

Synchronous handler

implicit threadpool

@get(sync_to_thread=True)

Dependency injection

Depends(fn)

dependencies={"name": Provide(fn)}

Application state

request.app.state

state: State + request.app.state

Lifespan

@asynccontextmanager

lifespan=[ctx_mgr]

JSON body

item: Item

data: Item

Form data

Form()

Body(media_type=URL_ENCODED)

File upload

UploadFile

UploadFile + Body(MULTI_PART)

Default POST status

200

201

Default DELETE status

200

204

Cookies

response.set_cookie

response_cookies=[Cookie(...)]

Templates

Jinja2Templates

Template() + TemplateConfig

HTTPException

HTTPException(400, ...)

HTTPException(status_code=400, ...)

Exception handler

@app.exception_handler

exception_handlers={Exc: handler}

Authentication

Depends chain

guard or auth middleware

Dependency overrides

app.dependency_overrides

factory + create_test_client

Middleware

BaseHTTPMiddleware

ASGIMiddleware subclass

Background task

BackgroundTasks injection

Response(background=BackgroundTask)

Streaming response

StreamingResponse

Stream(iterator)

WebSocket

@app.websocket("/ws")

@websocket_listener("/ws")

Test client

TestClient(app)

TestClient / create_test_client

OpenAPI customisation

per-decorator keywords

per-decorator keywords + OpenAPIConfig