Build a Python API with Litestar and SQLAlchemy

Litestar is a Python ASGI framework for building APIs with typed handlers, dependency injection, and generated OpenAPI schemas. This tutorial builds a small blog API with SQLite and async SQLAlchemy, keeping the database model separate from the objects clients send and receive.

Where Litestar fits

Previously called Starlite, Litestar offers data transfer objects (DTOs), database plugins, caching, templates, and static-file support. Its DTO system can control how application objects are accepted and serialized. The example below uses plain dataclasses and explicit conversion instead, so the API contract is easy to see.

If you are comparing Litestar with FastAPI, build a representative endpoint in each and examine validation, dependency scope, database integration, and deployment behavior. Architecture and team familiarity are more useful selection criteria than an unsupported performance ranking.

Set up the project

Use a virtual environment. This example targets Litestar 2.24 or later within the 2.x series and SQLAlchemy 2.x:

python -m venv .venv
source .venv/bin/activate
python -m pip install 'litestar[standard,sqlalchemy]>=2.24,<3' 'sqlalchemy>=2,<3' aiosqlite

Put the following Python blocks into app.py in the order shown. The SQLAlchemy plugin supplies a session for each request and handles its lifecycle; the handlers below explicitly commit writes.

Define the database model

from dataclasses import dataclass

from litestar import Litestar, delete, get, post, put
from litestar.di import NamedDependency
from litestar.exceptions import NotFoundException
from litestar.plugins.sqlalchemy import SQLAlchemyAsyncConfig, SQLAlchemyPlugin
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column


class Base(DeclarativeBase):
    pass


class Post(Base):
    __tablename__ = "posts"
    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str]
    body: Mapped[str]


config = SQLAlchemyAsyncConfig(
    connection_string="sqlite+aiosqlite:///blog.db",
    metadata=Base.metadata,
    create_all=True,
    session_dependency_key="session",
)
plugin = SQLAlchemyPlugin(config=config)

session_dependency_key="session" matches the handler parameter name. NamedDependency[AsyncSession] explicitly marks that parameter as an injected dependency, rather than request data. See the Litestar dependency injection reference.

create_all=True is convenient for this local example. Use database migrations for production schema changes: creating missing tables does not migrate existing ones.

Separate API inputs and outputs

The database assigns the ID. Clients provide only a title and body; responses contain all three fields.

@dataclass
class PostCreate:
    title: str
    body: str


PostUpdate = PostCreate  # PUT requires both fields.


@dataclass
class PostView:
    id: int
    title: str
    body: str


def as_view(post: Post) -> PostView:
    return PostView(id=post.id, title=post.title, body=post.body)

This separation lets the storage model evolve without automatically exposing every new database column. A Litestar DTO is another way to express that boundary; a DTO class is not itself the data object returned by a handler.

Create and read posts

@post("/posts")
async def create_post(
    data: PostCreate, session: NamedDependency[AsyncSession]
) -> PostView:
    post = Post(title=data.title, body=data.body)
    session.add(post)
    await session.commit()
    await session.refresh(post)
    return as_view(post)


@get("/posts")
async def list_posts(session: NamedDependency[AsyncSession]) -> list[PostView]:
    result = await session.execute(
        select(Post).order_by(Post.id.desc()).limit(100)
    )
    return [as_view(post) for post in result.scalars()]


@get("/posts/{pid:int}")
async def get_post(pid: int, session: NamedDependency[AsyncSession]) -> PostView:
    post = await session.get(Post, pid)
    if post is None:
        raise NotFoundException("Post not found")
    return as_view(post)

The create route returns 201 by default. The list is capped at 100 records, newest first; it is not a complete pagination design. A missing individual post returns 404.

The handlers await database I/O, allowing the event loop to serve other work while those operations wait. Declaring a handler async does not make blocking libraries or CPU-heavy processing nonblocking.

Update and delete posts

@put("/posts/{pid:int}")
async def update_post(
    pid: int, data: PostUpdate, session: NamedDependency[AsyncSession]
) -> PostView:
    post = await session.get(Post, pid)
    if post is None:
        raise NotFoundException("Post not found")
    post.title = data.title
    post.body = data.body
    await session.commit()
    await session.refresh(post)
    return as_view(post)


@delete("/posts/{pid:int}")
async def delete_post(pid: int, session: NamedDependency[AsyncSession]) -> None:
    post = await session.get(Post, pid)
    if post is None:
        raise NotFoundException("Post not found")
    await session.delete(post)
    await session.commit()


app = Litestar(
    plugins=[plugin],
    route_handlers=[create_post, list_posts, get_post, update_post, delete_post],
)

PUT requires both title and body. The delete handler returns no body and uses Litestar's default 204 status for DELETE. Start the local application:

litestar --app app:app run

Before exposing it beyond development, add field length and content constraints, authentication, per-object authorization, pagination, and appropriate transaction/error handling. The example deliberately has no access control.

Common questions

How should I choose between Litestar and FastAPI?

Compare the integrations you actually need, how request and response models are expressed, how dependencies are scoped, and the team's ability to maintain the result. Litestar's DTO and plugin systems are reasons to evaluate it; they do not establish that it is universally better for large applications.

Can Litestar serve templates and static files?

Yes. If the application is primarily server-rendered, compare those features and the surrounding ecosystem with Django or Flask before choosing. API support alone does not settle that decision.

What causes trouble with async SQLAlchemy?

Implicit I/O is a common source of errors: accessing an unloaded relationship may attempt a database query outside an awaited operation. Eager loading with selectinload() is one option; explicit refreshes and SQLAlchemy's async attribute support are others. Do not share an AsyncSession across concurrent tasks. The plugin manages request-scoped sessions, but any background work still needs its own session strategy. See SQLAlchemy's async guidance.

Follow a request beyond the API

Once the CRUD behavior is correct, instrument one request that reaches the database and inspect both spans and application logs. Guance's OpenTelemetry integration explains the receiving configuration. Confirm instrumentation compatibility and initialize it before the relevant framework or database libraries. This tutorial does not configure tracing or demonstrate a production deployment.

References and related reading

• Litestar SQLAlchemy plugin
• Litestar dependency injection
• Build an API with Flask (Chinese)
• Introduction to IPython (Chinese)

The CRUD example was exercised locally with SQLite. No load test, production database deployment, or Guance ingestion test was performed.