Async Database Access with SQLAlchemy 2.0 and FastAPI

Your API is only as fast as its slowest database round trip, and the database is usually the first line item on your infrastructure bill that scales with customers rather than with time. This guide shows you how to run async SQLAlchemy 2.0 against Postgres from an async-native FastAPI service the way a paying product needs it: one engine per process, a pool sized to your actual host, exactly one transaction per request, queries that do not silently issue two hundred round trips, and Alembic migrations that ship without a maintenance window. Part of the Scaling and Operating Production Python APIs guide.

Everything here assumes you have chosen a driver. If you have not, read asyncpg vs psycopg3 for FastAPI first — the code below uses asyncpg, which is the faster of the two for read-heavy JSON APIs, but every pattern on this page works with either.

Prerequisites

You need Python 3.11 or newer, Postgres 14 or newer, and four packages. Pin them; SQLAlchemy 1.4 and 2.0 differ enough in query style that a floating version will break your code at the worst moment.

pip install "sqlalchemy[asyncio]>=2.0.30" "asyncpg>=0.29" "alembic>=1.13" "fastapi>=0.115" "uvicorn[standard]>=0.30"

The connection URL must carry the postgresql+asyncpg scheme, not plain postgresql. If you paste the URL your host hands you into an async engine unchanged, SQLAlchemy loads the sync psycopg driver and every request raises MissingGreenlet the first time it touches the database. Normalise the scheme in code rather than in the dashboard, so the same environment variable feeds Alembic, your app, and your test suite:

export DATABASE_URL="postgresql+asyncpg://api:secret@localhost:5432/appdb"
export DB_POOL_SIZE="5"
export DB_MAX_OVERFLOW="5"
export DB_POOL_RECYCLE="1800"
export DB_STATEMENT_TIMEOUT_MS="5000"
export DB_ECHO="0"
export SERVERLESS="0"

One more assumption: your response models are Pydantic v2. Async SQLAlchemy punishes lazy attribute access during serialization, so the boundary between ORM object and response schema matters more than it does in a sync app. If your models are loose, tighten them with Pydantic v2 validation before you go further.

Step 1: Create exactly one engine, at process start

The engine is a heavyweight object that owns a connection pool. Create it once at module import or in the lifespan handler, never inside a request. A fresh engine per request opens a fresh TCP connection and a fresh TLS handshake every time, which typically adds 20–40 ms to every call and exhausts the server's connection slots under any real concurrency.

Keep the engine, the sessionmaker, and the declarative base in one small module. The AsyncAttrs mixin on the base is not optional in my opinion — it gives you await obj.awaitable_attrs.<relationship> as an escape hatch when you genuinely need a lazy load, instead of a runtime crash.

import os
from sqlalchemy.ext.asyncio import (
    AsyncAttrs,
    AsyncEngine,
    AsyncSession,
    async_sessionmaker,
    create_async_engine,
)
from sqlalchemy.orm import DeclarativeBase
from sqlalchemy.pool import NullPool


class Base(AsyncAttrs, DeclarativeBase):
    """Declarative base with awaitable attribute access."""


def _url() -> str:
    raw = os.environ["DATABASE_URL"]
    # Hosts hand out sync URLs; force the async driver.
    if raw.startswith("postgresql://"):
        raw = raw.replace("postgresql://", "postgresql+asyncpg://", 1)
    return raw


def build_engine() -> AsyncEngine:
    serverless = os.getenv("SERVERLESS", "0") == "1"
    timeout_ms = os.getenv("DB_STATEMENT_TIMEOUT_MS", "5000")
    connect_args = {
        "server_settings": {
            "application_name": os.getenv("APP_NAME", "api"),
            "statement_timeout": timeout_ms,
        },
        "timeout": float(os.getenv("DB_CONNECT_TIMEOUT", "10")),
    }

    match serverless:
        case True:
            return create_async_engine(
                _url(),
                poolclass=NullPool,
                connect_args={**connect_args, "statement_cache_size": 0},
                echo=os.getenv("DB_ECHO", "0") == "1",
            )
        case False:
            return create_async_engine(
                _url(),
                pool_size=int(os.getenv("DB_POOL_SIZE", "5")),
                max_overflow=int(os.getenv("DB_MAX_OVERFLOW", "5")),
                pool_timeout=float(os.getenv("DB_POOL_TIMEOUT", "10")),
                pool_recycle=int(os.getenv("DB_POOL_RECYCLE", "1800")),
                pool_pre_ping=True,
                echo=os.getenv("DB_ECHO", "0") == "1",
            )


engine = build_engine()

SessionLocal = async_sessionmaker(
    engine,
    expire_on_commit=False,
    autoflush=False,
)

Two settings there earn their place. pool_pre_ping=True costs a trivial round trip when a connection is checked out and saves you from the single most common production error in this stack: a managed Postgres or a load balancer silently closing idle connections after five or ten minutes, leaving stale sockets in your pool that fail the next request with ConnectionDoesNotExistError. pool_recycle=1800 handles the same problem proactively by discarding connections older than half an hour.

expire_on_commit=False matters even more. By default SQLAlchemy expires every loaded attribute after a commit, so the next attribute access triggers a refresh query. In an async app that refresh happens outside a greenlet context during response serialization and raises MissingGreenlet instead of quietly reloading. Turn it off and treat objects returned from a committed transaction as immutable snapshots.

Dispose the engine on shutdown so in-flight connections close cleanly rather than being reaped by Postgres later:

from contextlib import asynccontextmanager
from fastapi import FastAPI


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


app = FastAPI(lifespan=lifespan)

Step 2: Size the pool for the host you actually run on

Pool sizing is arithmetic, not taste. Your total connection demand is replicas × worker processes × (pool_size + max_overflow). Compare that to your Postgres max_connections minus the slots reserved for superusers, your migration job, your analytics tool, and any background worker fleet. If the total exceeds the ceiling, you do not get slow requests — you get FATAL: sorry, too many clients already, which surfaces as 500s across every endpoint simultaneously.

Work backwards. A typical managed Postgres starter plan gives you 100 connections. Reserve 20 for humans, migrations, and Celery workers, leaving 80 for the API. Two replicas of four Uvicorn workers is eight processes, so each process gets ten: pool_size=5, max_overflow=5. Steady-state traffic sits inside the five pooled connections and bursts borrow up to five more, which are closed again when they go idle.

Serverless inverts the logic. On a platform that freezes or destroys the process between invocations — Vercel functions, Lambda, Cloud Run scaled to zero — an in-process pool is worthless and actively harmful. Each cold instance opens its own pooled connections, then gets frozen while still holding them. Fifty concurrent instances with a five-connection pool means 250 half-dead connections against a 100-slot server. Use NullPool so every request opens and closes a single connection, and put a real pooler in front: PgBouncer in transaction mode, Supabase's pooler port, or Neon's pooled endpoint. The pooler multiplexes thousands of short client connections onto a few dozen server connections, which is exactly the job an in-process pool cannot do across process boundaries.

One caveat when you go through PgBouncer in transaction mode: asyncpg's prepared statement cache breaks, because a prepared statement created on one server connection is not visible on the next. Set statement_cache_size=0 in connect_args, as the serverless branch above does. Skip that and you get sporadic InvalidSQLStatementNameError under load — a bug that never reproduces on your laptop. The full diagnosis lives in fixing connection pool exhaustion.

Step 3: Scope one transaction per request

The cleanest transaction boundary in a web API is the request itself. Open a session, begin a transaction, run the handler, commit on success, roll back on any exception. A FastAPI dependency with yield expresses this in six lines, and it means no handler ever calls commit() by hand — which in turn means no half-written record survives a mid-handler exception.

from collections.abc import AsyncIterator
from typing import Annotated
from fastapi import Depends
from sqlalchemy.ext.asyncio import AsyncSession


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


DbSession = Annotated[AsyncSession, Depends(get_session)]

session.begin() commits when the block exits normally and rolls back when an exception escapes, including HTTPException. That last detail is the point: raise HTTPException(status_code=409) halfway through a multi-table write and the whole thing unwinds. You get atomic endpoints for free.

The cost of this pattern is that a connection stays checked out for the entire handler, including any HTTP call you make to Stripe or an LLM provider in the middle of it. That is the fastest way to exhaust a pool: ten concurrent requests each holding a connection while waiting three seconds on a third-party API will stall every other request behind pool_timeout. Never call an external service inside a database transaction. Read what you need, commit, then make the call, then open a second short transaction to record the result — or push the call into a background job entirely.

Here is a handler using the dependency, with a write path that stays inside one transaction:

import os
import uuid
from fastapi import HTTPException
from sqlalchemy import select
from sqlalchemy.exc import IntegrityError


@app.post("/v1/projects", status_code=201)
async def create_project(payload: ProjectIn, db: DbSession) -> ProjectOut:
    existing = await db.scalar(
        select(Project).where(
            Project.owner_id == payload.owner_id,
            Project.slug == payload.slug,
        )
    )
    if existing is not None:
        raise HTTPException(status_code=409, detail="slug already in use")

    project = Project(
        id=uuid.uuid4(),
        owner_id=payload.owner_id,
        slug=payload.slug,
        name=payload.name,
    )
    db.add(project)
    try:
        await db.flush()
    except IntegrityError as exc:
        raise HTTPException(status_code=409, detail="slug already in use") from exc
    return ProjectOut.model_validate(project, from_attributes=True)

Note the flush() rather than commit(). Flush pushes the INSERT so the unique constraint fires and you can turn it into a clean 409, while the dependency still owns the commit. Keep the unique index in the database as well as the pre-check — the pre-check is a nicety for the error message, the index is what actually prevents duplicates under concurrency.

Step 4: Kill N+1 queries before they reach production

N+1 is the default behaviour of a relational ORM and the single largest source of latency in ORM-backed APIs. You select 200 projects, then serialize each one's owner, and SQLAlchemy issues 200 extra queries. On a local database each is 0.9 ms and you never notice. Across a network to a managed Postgres each is 4 ms, and your p95 for that endpoint jumps from 15 ms to 800 ms — with 200x the query volume billed against your connection budget.

In async SQLAlchemy the failure mode is at least loud: a lazy load during serialization raises MissingGreenlet rather than silently issuing queries. Lean into that. Declare relationships with lazy="raise_on_sql" so any accidental lazy load blows up in your test suite instead of degrading in production.

from datetime import datetime
import uuid
from sqlalchemy import ForeignKey, String, func
from sqlalchemy.orm import Mapped, mapped_column, relationship
from sqlalchemy.dialects.postgresql import UUID


class Owner(Base):
    __tablename__ = "owners"

    id: Mapped[uuid.UUID] = mapped_column(UUID(as_uuid=True), primary_key=True)
    email: Mapped[str] = mapped_column(String(320), unique=True)
    projects: Mapped[list["Project"]] = relationship(
        back_populates="owner", lazy="raise_on_sql"
    )


class Project(Base):
    __tablename__ = "projects"

    id: Mapped[uuid.UUID] = mapped_column(UUID(as_uuid=True), primary_key=True)
    owner_id: Mapped[uuid.UUID] = mapped_column(
        ForeignKey("owners.id", ondelete="CASCADE"), index=True
    )
    slug: Mapped[str] = mapped_column(String(64))
    name: Mapped[str] = mapped_column(String(200))
    created_at: Mapped[datetime] = mapped_column(server_default=func.now())
    owner: Mapped[Owner] = relationship(back_populates="projects", lazy="raise_on_sql")

Now every query states its loading strategy explicitly. Use selectinload for collections and joinedload for many-to-one scalars. selectinload issues a second query with an IN clause, which stays fast and avoids the row multiplication that a join across a collection causes. joinedload folds a single related row into the same query, which is ideal for project.owner.

from sqlalchemy.orm import joinedload, selectinload


async def list_projects(db: AsyncSession, owner_id: uuid.UUID, limit: int = 200):
    stmt = (
        select(Project)
        .where(Project.owner_id == owner_id)
        .options(joinedload(Project.owner))
        .order_by(Project.created_at.desc())
        .limit(limit)
    )
    result = await db.scalars(stmt)
    return result.all()


async def list_owners_with_projects(db: AsyncSession, limit: int = 50):
    stmt = (
        select(Owner)
        .options(selectinload(Owner.projects))
        .order_by(Owner.email)
        .limit(limit)
    )
    result = await db.scalars(stmt)
    return result.unique().all()

The numbers below come from a 200-row listing against a managed Postgres roughly 4 ms of round trip away. The shape holds on any remote database: the fix is not micro-optimisation, it is deleting 200 network hops.

Two rules keep this honest long term. First, call .unique() on any result that used joinedload against a collection, or SQLAlchemy raises rather than hand you duplicated parents. Second, add a per-request query counter to your logs so a regression shows up in a dashboard rather than in a support ticket; the monitoring and logging setup page covers wiring it into structured logs. If a listing endpoint stays hot and its data changes rarely, the next lever is caching responses with Redis so the query does not run at all.

Step 5: Run Alembic migrations against the async engine

Alembic ships an async template. Use it — hand-rolling asyncio.run() inside env.py is how people end up with migrations that hang on a closed loop.

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

Point Alembic at the same environment variable your app reads, so there is exactly one source of truth for the connection string. Edit migrations/env.py:

import asyncio
import os
from alembic import context
from sqlalchemy.ext.asyncio import async_engine_from_config
from sqlalchemy import pool

from app.db import Base, _url
from app import models  # noqa: F401 — registers tables on Base.metadata

config = context.config
config.set_main_option("sqlalchemy.url", _url())
target_metadata = Base.metadata


def do_run_migrations(connection) -> None:
    context.configure(
        connection=connection,
        target_metadata=target_metadata,
        compare_type=True,
        compare_server_default=True,
    )
    with context.begin_transaction():
        context.run_migrations()


async def run_async_migrations() -> None:
    connectable = async_engine_from_config(
        config.get_section(config.config_ini_section, {}),
        prefix="sqlalchemy.",
        poolclass=pool.NullPool,
    )
    async with connectable.connect() as connection:
        await connection.run_sync(do_run_migrations)
    await connectable.dispose()


asyncio.run(run_async_migrations())

compare_type=True and compare_server_default=True are worth turning on: without them autogenerate misses a String(64) widening to String(200) and you ship a migration that does nothing. Always read the generated file before committing it — autogenerate is a first draft, not an oracle. It routinely wants to drop indexes it does not know about and rename tables it should leave alone.

The dangerous migrations are the ones that take a lock. ALTER TABLE ... ADD COLUMN NOT NULL DEFAULT rewrites the whole table on older Postgres versions and holds an ACCESS EXCLUSIVE lock the entire time; on a million-row table that is a multi-minute outage. Guard every migration with a lock timeout so it fails fast instead of queueing every query behind it, and build indexes concurrently:

from alembic import op
import sqlalchemy as sa


def upgrade() -> None:
    op.execute("SET lock_timeout = '3s'")
    op.add_column("projects", sa.Column("archived_at", sa.DateTime(timezone=True)))
    with op.get_context().autocommit_block():
        op.create_index(
            "ix_projects_archived_at",
            "projects",
            ["archived_at"],
            postgresql_concurrently=True,
        )

For anything that changes an existing column's meaning, use expand and contract: add the new column as nullable, deploy code that writes both old and new, backfill in batches, deploy code that reads only the new column, then drop the old one in a later release. Each step is independently reversible and none of them requires the app and the schema to change at the same instant — which is what makes zero-downtime deploys possible at all.

Configuration reference

The two values worth tuning per deployment are DB_POOL_SIZE and DB_STATEMENT_TIMEOUT_MS. Pool size follows the arithmetic from Step 2 and should change whenever you add replicas or workers. The statement timeout is a safety valve: it caps how long one bad query can pin a connection, and Postgres enforces it server-side, so it works even when your application is too busy to cancel the query itself. Set it once per role, not per query, and give background jobs a separate database role with a much longer timeout.

Gotchas and failure modes

Lazy loading during response serialization. Symptom: sqlalchemy.exc.MissingGreenlet: greenlet_spawn has not been called raised from inside Pydantic serialization, only on endpoints that return nested data, and never in the unit tests that hit the repository layer directly. Cause: an attribute that was not eagerly loaded gets touched after the session closed. Fix: declare lazy="raise_on_sql" on every relationship and add the explicit selectinload/joinedload option to the query. If you need one lazy load deliberately, use await obj.awaitable_attrs.owner while the session is still open.

A sync driver behind an async engine. Symptom: every request fails immediately with InvalidRequestError: The asyncio extension requires an async driver. Cause: DATABASE_URL starts with postgresql:// because that is what the hosting dashboard emits. Fix: normalise the scheme in code as _url() does above, and add a startup assertion so a misconfigured environment fails at boot rather than on first traffic.

Holding a connection across an HTTP call. Symptom: TimeoutError: QueuePool limit of size 5 overflow 5 reached under load that your database barely notices — CPU at 8%, connections pegged. Cause: a handler awaits Stripe or an LLM inside the request transaction. Fix: commit before the outbound call and open a second transaction to persist the result, or hand the whole thing to a background worker.

Sharing one session across concurrent tasks. Symptom: InterfaceError: cannot perform operation: another operation is in progress, or rows from one request appearing in another. Cause: passing a single AsyncSession into asyncio.gather(). A session is not concurrency-safe. Fix: give each task its own session from SessionLocal(), or run the queries sequentially — usually they are fast enough that concurrency buys nothing.

Autogenerate producing a destructive migration. Symptom: a review-less alembic upgrade head drops an index or a column that a table needs, because the model file was mid-refactor when the revision was generated. Fix: read every generated migration, run it against a restored copy of production data in CI, and never let the deploy pipeline generate migrations — it may only apply committed ones.

Verification

Prove the wiring end to end before you trust it. Add a readiness endpoint that actually touches the database and reports pool state, then hit it while the app is running:

from sqlalchemy import text


@app.get("/healthz/db")
async def db_health(db: DbSession) -> dict[str, object]:
    value = await db.scalar(text("SELECT 1"))
    pool = engine.pool
    return {
        "ok": value == 1,
        "checked_out": pool.checkedout(),
        "size": pool.size(),
        "overflow": pool.overflow(),
    }

curl -s "${API_BASE_URL}/healthz/db" | python -m json.tool

Expect {"ok": true, "checked_out": 1, "size": 5, "overflow": -4} on an idle process — checked_out is 1 because the request itself holds a connection, and a negative overflow simply means the pool has not filled yet. If checked_out stays near size + max_overflow after traffic drains, you are leaking sessions somewhere.

For the query count, assert it in a test rather than eyeballing logs. A before_cursor_execute listener counts statements, and the test fails if the endpoint regresses to N+1:

import pytest
from sqlalchemy import event


@pytest.mark.asyncio
async def test_listing_is_not_n_plus_one(client, engine):
    count = 0

    def _count(conn, cursor, statement, params, ctx, many):
        nonlocal count
        count += 1

    sync_engine = engine.sync_engine
    event.listen(sync_engine, "before_cursor_execute", _count)
    try:
        response = await client.get("/v1/owners?limit=50")
    finally:
        event.remove(sync_engine, "before_cursor_execute", _count)

    assert response.status_code == 200
    assert count <= 3, f"expected <= 3 queries, issued {count}"

That test is the highest-value one on this page; wire it into the suite described in testing Python APIs with pytest and an N+1 regression can never reach a customer.

Cost and performance at scale

Database access is where an API's unit economics are decided. At a million requests a month — roughly 0.4 requests per second average, with peaks perhaps ten times that — a well-loaded endpoint issuing one or two queries at 4 ms each keeps p95 latency under 40 ms and runs comfortably on a $20–$40 managed Postgres and two small API containers. The same endpoint with an N+1 pattern issues 200 million queries a month instead of two million, which does not merely add latency: it pushes you onto a larger database plan for connection headroom and IOPS, adds a second or third API replica to absorb the latency, and turns a $60 monthly infrastructure line into $300 or more. Fixing the loading strategy is a fifteen-minute change that recovers the entire difference, and it compounds — every customer you add rides the cheaper curve. On serverless the arithmetic is harsher still, because you also pay for wall-clock execution time: 800 ms of waiting on N+1 queries is 800 ms of billed compute per request, so the same fix cuts both your database bill and your function bill at once. Before you set prices, run the numbers for your own endpoints with calculating cost per API request; a query-count budget per endpoint is the cheapest performance guardrail you will ever add.

FAQ

How much does a Postgres-backed API cost to run at a million requests a month? Around $60 a month all in if you keep query counts low: roughly $25 for a small managed Postgres, $15–$30 for two API containers, and near-zero egress for JSON. The variable that moves the number is queries per request, not requests per second — an N+1 listing endpoint can multiply the database line by five without any change in traffic.

Should I use an external pooler like PgBouncer, or is the SQLAlchemy pool enough? On long-lived containers with a handful of processes, the SQLAlchemy pool is enough and adds one less moving part. Add PgBouncer in transaction mode the moment you go serverless, exceed roughly 50 total processes, or start seeing connection-limit errors during deploys when old and new instances overlap. Remember to set statement_cache_size=0 when you do.

What is the migration risk when I change a table that customers depend on? The risk is lock duration, not the schema change itself. A migration that holds ACCESS EXCLUSIVE on a large table queues every query behind it and reads to your customers as a full outage. Set lock_timeout, build indexes concurrently, and split anything semantic into expand-and-contract steps so no single release requires the code and schema to flip together.

Do I need async at all, or would sync SQLAlchemy with more workers be cheaper? If your endpoints do one fast query and return, sync plus more worker processes is genuinely simpler and costs about the same. Async pays for itself when a request waits on several I/O operations — a database read plus two upstream API calls — because one worker handles hundreds of concurrent waits instead of one. Most commercial APIs cross that line quickly, which is why the async stack is the better default.

How do I keep read-heavy analytics queries from starving my API pool? Give them their own database role, their own engine, and their own pool — ideally pointed at a read replica. Analytics scans hold connections for seconds; API queries hold them for milliseconds. Mixing them in one pool means one dashboard refresh can time out live traffic. The same separation applies to the writes behind logging API usage events to Postgres.

Related

Same track:

• asyncpg vs psycopg3 for FastAPI — pick the driver before you tune anything else.
• Fixing Connection Pool Exhaustion — the diagnosis playbook when QueuePool limit errors start firing.
• Caching Python API Responses with Redis — stop running the query at all for hot, slow-changing reads.
• Testing Python APIs with pytest — where the query-count assertion above belongs.

Other tracks:

• Zero-Downtime Deploys for Python APIs — how migrations and rollouts stay in step.