FastAPI can handle many concurrent requests efficiently, but an async def endpoint does not make a synchronous database driver asynchronous. To avoid blocking the event loop, the entire database path—from PostgreSQL driver to SQLAlchemy session—must support async I/O.

This guide builds a small FastAPI API using SQLAlchemy 2.0, Psycopg 3, and PostgreSQL. It focuses on the production details that cause the most trouble: one session per request, bounded connection pools, transactions, cleanup, migrations, and tests.

When async database access helps

Async I/O is useful when requests spend substantial time waiting for PostgreSQL or other network services. While one coroutine waits, the event loop can serve another request. It does not make expensive Python calculations faster, and it cannot compensate for slow SQL or missing indexes.

FastAPI recommends async def when a library exposes awaitable calls. SQLAlchemy's async extension provides those calls while preserving the ORM and transaction model. Review the official guides for FastAPI concurrency and SQLAlchemy asyncio before mixing synchronous and asynchronous code.

Install the dependencies

python -m venv .venv
source .venv/bin/activate

pip install fastapi "uvicorn[standard]" \
  "sqlalchemy[asyncio]" "psycopg[binary]" \
  pydantic-settings alembic

Use an environment variable for the connection URL:

DATABASE_URL=postgresql+psycopg://app_user:strong_password@127.0.0.1:5432/app_db

Do not commit production credentials. In CI and production, inject them from the platform's secret manager or protected environment.

Create a bounded async engine

Create app/database.py:

from collections.abc import AsyncIterator

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

DATABASE_URL = "postgresql+psycopg://app_user:strong_password@localhost/app_db"

engine = create_async_engine(
    DATABASE_URL,
    pool_size=10,
    max_overflow=10,
    pool_timeout=30,
    pool_pre_ping=True,
)

SessionFactory = async_sessionmaker(
    engine,
    class_=AsyncSession,
    expire_on_commit=False,
)

async def get_session() -> AsyncIterator[AsyncSession]:
    async with SessionFactory() as session:
        yield session

pool_size is a per-process limit. If four application workers each allow ten persistent connections plus ten overflow connections, the theoretical peak is 80. Size the combined total below PostgreSQL's available connection budget and leave capacity for migrations, monitoring, and administration.

pool_pre_ping checks connections before use and helps recover from server-side disconnects. It does not replace sensible timeouts, health checks, or a managed connection pooler when the deployment has many application processes.

Define an ORM model and schemas

from datetime import datetime

from pydantic import BaseModel, ConfigDict
from sqlalchemy import DateTime, String, func
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column

class Base(DeclarativeBase):
    pass

class Project(Base):
    __tablename__ = "projects"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(120), unique=True, index=True)
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), server_default=func.now()
    )

class ProjectCreate(BaseModel):
    name: str

class ProjectRead(BaseModel):
    model_config = ConfigDict(from_attributes=True)

    id: int
    name: str
    created_at: datetime

Keep API schemas separate from ORM models. This prevents accidental exposure of internal columns and gives request validation an independent lifecycle.

Use one AsyncSession per request

An AsyncSession represents mutable transaction state and must not be shared between concurrent tasks. Inject a new session into each request:

from typing import Annotated

from fastapi import Depends, FastAPI, HTTPException, status
from sqlalchemy import select
from sqlalchemy.exc import IntegrityError
from sqlalchemy.ext.asyncio import AsyncSession

app = FastAPI()
DbSession = Annotated[AsyncSession, Depends(get_session)]

@app.post("/projects", response_model=ProjectRead, status_code=201)
async def create_project(payload: ProjectCreate, session: DbSession):
    project = Project(name=payload.name.strip())
    session.add(project)

    try:
        await session.commit()
    except IntegrityError:
        await session.rollback()
        raise HTTPException(
            status_code=status.HTTP_409_CONFLICT,
            detail="Project name already exists",
        )

    await session.refresh(project)
    return project

@app.get("/projects", response_model=list[ProjectRead])
async def list_projects(session: DbSession):
    result = await session.execute(
        select(Project).order_by(Project.created_at.desc()).limit(100)
    )
    return result.scalars().all()

Always rollback after a failed commit before reusing the session. Convert expected constraint failures into stable API errors, but log unexpected database exceptions with a request correlation ID.

Make transaction boundaries explicit

For operations containing several writes, use a transaction context:

async with SessionFactory.begin() as session:
    session.add(project)
    session.add(AuditEvent(action="project.created"))

The context commits if the block succeeds and rolls back if it raises. Keep transactions short: do not hold a database transaction open while calling an email provider, object storage API, or other slow network dependency.

Dispose connections during shutdown

FastAPI's lifespan hook is the right place to release the pool:

from contextlib import asynccontextmanager
from fastapi import FastAPI

@asynccontextmanager
async def lifespan(app: FastAPI):
    yield
    await engine.dispose()

app = FastAPI(lifespan=lifespan)

This matters in tests, reloads, and graceful production shutdowns. FastAPI documents lifespan as the recommended way to manage startup and shutdown resources.

Use Alembic for schema changes

Do not run Base.metadata.create_all() every time production starts. Initialize Alembic, point its metadata to Base.metadata, and create reviewed migrations:

alembic init migrations
alembic revision --autogenerate -m "create projects"
alembic upgrade head

Run migrations once per deployment, not independently in every web worker. Review generated SQL, especially for type changes, unique constraints, and operations that may lock large tables.

Avoid common async mistakes

  • Do not call synchronous database clients from an async endpoint.
  • Do not create a global AsyncSession.
  • Do not spawn concurrent tasks that share one session.
  • Avoid implicit lazy loading; load relationships explicitly.
  • Limit query results and add indexes based on measured query plans.
  • Count pool capacity across all server workers.
  • Set statement and request timeouts so stalled work eventually ends.

Async FastAPI and PostgreSQL work well together when every layer is truly asynchronous and connection usage is bounded. Treat the session as request-scoped transaction state, keep transactions short, migrate schemas separately, dispose the engine cleanly, and measure the database rather than assuming async code will fix inefficient queries.