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
Application (
Litestar)Router/Controller(these are on the same level and can be arbitrarily nested)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])
See also
Routers#
A few small differences between Litestar’s Router and FastAPI / Starlette’s:
A Litestar
Routeris not itself an ASGI application. It groups handlers and options. Routers are “flattened” into simple handlers during registrationA Litestar
Routerdoes not expose decoratorsA Litestar
Routerdoes 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.
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.
See also
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:
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
Construct a fresh application per test with
create_test_client(), passing the overridden dependency through the samedependencies=keyword the production application usesPatch 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},
)
See also
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()])
See also
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.websocketwebsocket_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 |
|
|
Synchronous handler |
implicit threadpool |
|
Dependency injection |
|
|
Application state |
|
|
Lifespan |
|
|
JSON body |
|
|
Form data |
|
|
File upload |
|
|
Default POST status |
|
|
Default DELETE status |
|
|
Cookies |
|
|
Templates |
|
|
HTTPException |
|
|
Exception handler |
|
|
Authentication |
|
guard or auth middleware |
Dependency overrides |
|
factory + |
Middleware |
|
|
Background task |
|
|
Streaming response |
|
|
WebSocket |
|
|
Test client |
|
|
OpenAPI customisation |
per-decorator keywords |
per-decorator keywords + OpenAPIConfig |