FastAPI handles HTTP requests, while SQLAlchemy manages database operations. Together, they provide a practical way to build database-driven APIs using Python and PostgreSQL.

This guide covers asynchronous database configuration, ORM models, dependency injection, and database migrations with Alembic.

1. Install Dependencies

pip install fastapi uvicorn "sqlalchemy[asyncio]" asyncpg alembic

• SQLAlchemy: ORM and database toolkit.
• asyncpg: Asynchronous PostgreSQL driver.
• Alembic: Database schema migration tool.
• Uvicorn: ASGI server for running FastAPI.

2. Configure the Database

Create app/database.py to configure the database engine, session factory, and dependency.

import os
from collections.abc import AsyncGenerator

from sqlalchemy.ext.asyncio import (
    AsyncSession,
    async_sessionmaker,
    create_async_engine,
)

DATABASE_URL = os.environ["DATABASE_URL"]

engine = create_async_engine(DATABASE_URL)

AsyncSessionLocal = async_sessionmaker(
    engine,
    expire_on_commit=False,
)


async def get_db() -> AsyncGenerator[AsyncSession, None]:
    async with AsyncSessionLocal() as session:
        yield session

Configure DATABASE_URL through your environment, for example:

postgresql+asyncpg://user:password@localhost:5432/plant_shop

Key components:

• Engine: Manages database connectivity and connection pooling.
• Session factory: Creates new database sessions.
• get_db(): Provides a session to the request and closes it when the dependency finishes.

The session context manager handles cleanup, but it does not automatically commit transactions.

3. Define a SQLAlchemy Model

A SQLAlchemy model maps a Python class to a database table.

Create app/models.py:

from decimal import Decimal

from sqlalchemy import Integer, Numeric, String
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column


class Base(DeclarativeBase):
    pass


class Plant(Base):
    __tablename__ = "plants"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(100))
    price: Mapped[Decimal] = mapped_column(Numeric(10, 2))
    stock: Mapped[int] = mapped_column(Integer, default=0)

Mapped and mapped_column() provide SQLAlchemy 2.x’s typed ORM mapping syntax.

Use Decimal with Numeric for monetary values to avoid floating-point precision issues.

Keep SQLAlchemy models separate from Pydantic schemas. ORM models represent database tables, while Pydantic schemas validate API inputs and define response structures.

4. Query the Database with FastAPI

Use Depends() to inject a database session into an endpoint.

Create app/main.py:

from fastapi import Depends, FastAPI
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from app.database import get_db
from app.models import Plant

app = FastAPI()


@app.get("/plants")
async def get_plants(
    db: AsyncSession = Depends(get_db),
):
    result = await db.execute(select(Plant))
    return result.scalars().all()

Here’s what happens:

1. FastAPI resolves get_db() and provides an AsyncSession.
2. SQLAlchemy executes the query asynchronously.
3. scalars().all() extracts the returned Plant objects.
4. FastAPI serializes the returned data into an HTTP response.

For production APIs, define a Pydantic response schema with response_model rather than returning ORM objects directly.

5. Insert and Commit Records

Use the session to add a record and persist it to the database.

plant = Plant(
    name="Monstera",
    price=Decimal("250000.00"),
    stock=10,
)

db.add(plant)
await db.commit()
await db.refresh(plant)

• db.add(): Adds the object to the session.
• await db.commit(): Commits the transaction.
• await db.refresh(): Reloads database-generated values, such as the assigned ID.

Unlike execute() and commit(), add() is synchronous because it stages the object in the session rather than performing an immediate database operation.

In production code, handle failures and roll back an active transaction when necessary.

6. Manage Database Migrations with Alembic

Database models can change as an application evolves. Alembic tracks schema changes through versioned migration scripts.

Initialize an asynchronous migration environment:

alembic init -t async migrations

Configure Alembic’s target_metadata to use Base.metadata and ensure the migration environment can import your models.

Generate and apply a migration:

alembic revision --autogenerate -m "create plants"
alembic upgrade head

• revision --autogenerate: Generates a migration based on differences between your models and database schema.
• upgrade head: Applies migrations up to the latest revision.

Always review autogenerated migrations before applying them. Alembic does not detect every schema change automatically, and model definitions alone do not create or update database tables.

Key Takeaways

• Engine: Manages database connectivity and pooling.
• AsyncSession: Handles database operations and transactions.
• Dependency injection: Provides a session to each request.
• SQLAlchemy models: Map Python classes to database tables.
• Pydantic schemas: Define and validate API data.
• Alembic: Tracks and applies database schema changes.

Combining FastAPI, SQLAlchemy 2.x, and Alembic provides a maintainable foundation for asynchronous PostgreSQL APIs.