Async Database Access

Published by Programming Line in FastAPI. Last updated 6th September 2026.

Async database access in FastAPI means the request handler can pause while the database driver waits on the network, letting the event loop run other requests in the same worker. The outcome is not faster SQL. The outcome is better concurrency for I/O-bound endpoints when every part of the call path is awaitable: the route, dependencies, ORM session, driver, and transaction helpers.

This lesson belongs in the Databases and Transactions section because async access changes how connection lifetime and transaction lifetime are owned. A FastAPI endpoint should acquire a connection or session late, release it reliably, and keep transaction boundaries visible enough that a failed request cannot leave half-applied data.

Purpose and Outcome

Use async database access when a service spends meaningful time waiting for a database and must serve many overlapping requests. A typical path is: an ASGI server receives a request, FastAPI resolves dependencies, the endpoint awaits a query, the async driver writes bytes to a socket, and the event loop resumes another coroutine until the database response arrives.

If one request performs a slow query, unrelated requests can still be processed, subject to CPU, pool size, and database capacity. If an async def endpoint calls a synchronous PostgreSQL client, the event loop is pinned until that call returns, and the application has far less concurrency than the code suggests.

How the Mechanism Works

FastAPI runs on ASGI. An async def route returns a coroutine. When that coroutine executes await session.execute(...) or await connection.fetchval(...), control returns to the event loop until the awaited operation is ready. The database driver must be built for asyncio, such as asyncpg directly or SQLAlchemy’s async layer with an async dialect like postgresql+asyncpg.

A connection pool is central. Opening a database connection is expensive, so applications create a pool at startup or create an engine that owns one. Each request checks out a connection, uses it briefly, then returns it. Pool size is a concurrency limit. If twenty requests reach the database simultaneously and the pool has five connections, fifteen wait for checkout. That wait is non-blocking to the event loop, but it is real latency for those requests.

With SQLAlchemy, AsyncEngine owns the pool and dialect, async_sessionmaker creates sessions, and AsyncSession tracks ORM identity state and pending changes. The session is not global and should not be shared across concurrent requests. A common FastAPI dependency yields one session per request and closes it after the response or exception.

Transactions are separate from sessions. A session may exist before a transaction starts. For endpoints that change multiple rows, prefer async with session.begin() so commit and rollback behavior is tied to one clear region. If an exception leaves the block, SQLAlchemy rolls back. If the block completes, it commits.

API Anatomy

Piece	Role	Design concern
`async def` endpoint	Allows FastAPI to await request work	Do not call blocking database code inside it
Async driver	Performs socket I/O without blocking the loop	Use a matching URL scheme such as `postgresql+asyncpg`
Pool	Reuses database connections	Size it below the database connection budget
`AsyncSession`	Unit of work for ORM queries and changes	Create per request or per job
`session.begin()`	Commit or rollback boundary	Keep unrelated network calls outside it

Example 1: Direct Driver Health Check

This example uses asyncpg without an ORM. Startup creates a pool and shutdown closes it. The endpoint checks out a connection, awaits select 1, and returns a deterministic body: {"database":"ok","result":"1"}.

import asyncpg
from fastapi import FastAPI

app = FastAPI()

@app.on_event("startup")
async def open_pool() -> None:
    app.state.pool = await asyncpg.create_pool(
        "postgresql://app:secret@localhost:5432/appdb",
        min_size=1,
        max_size=10,
    )

@app.on_event("shutdown")
async def close_pool() -> None:
    await app.state.pool.close()

@app.get("/health/database")
async def database_health() -> dict[str, str]:
    async with app.state.pool.acquire() as connection:
        value = await connection.fetchval("select 1")
    return {"database": "ok", "result": str(value)}

This is the smallest useful async database path: pool acquisition, one awaited query, and connection release through async with. Direct drivers are fast and explicit, but larger applications often need migrations, query composition, and model mapping.

Example 2: Request-Scoped AsyncSession

The next example uses SQLAlchemy’s async API. The engine string selects an async dialect, the session factory disables expiration on commit so refreshed values remain easy to serialize, and get_session yields a session for the current request. Creating a note returns 201 with a body like {"id":1,"title":"async path"}. Reading a missing note returns 404.

from collections.abc import AsyncIterator

from fastapi import Depends, FastAPI, HTTPException
from pydantic import BaseModel
from sqlalchemy import Integer, String, select
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column

DATABASE_URL = "postgresql+asyncpg://app:secret@localhost:5432/appdb"
engine = create_async_engine(DATABASE_URL, pool_size=5, max_overflow=10)
SessionLocal = async_sessionmaker(engine, expire_on_commit=False)
app = FastAPI()

class Base(DeclarativeBase):
    pass

class Note(Base):
    __tablename__ = "notes"
    id: Mapped[int] = mapped_column(Integer, primary_key=True)
    title: Mapped[str] = mapped_column(String(120), nullable=False)

class NoteCreate(BaseModel):
    title: str

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

@app.post("/notes", status_code=201)
async def create_note(
    payload: NoteCreate,
    session: AsyncSession = Depends(get_session),
) -> dict[str, int | str]:
    note = Note(title=payload.title)
    session.add(note)
    await session.commit()
    await session.refresh(note)
    return {"id": note.id, "title": note.title}

@app.get("/notes/{note_id}")
async def read_note(
    note_id: int,
    session: AsyncSession = Depends(get_session),
) -> dict[str, int | str]:
    result = await session.execute(select(Note).where(Note.id == note_id))
    note = result.scalar_one_or_none()
    if note is None:
        raise HTTPException(status_code=404, detail="Note not found")
    return {"id": note.id, "title": note.title}

session.add() is not awaited because it only changes in-memory ORM state. commit(), refresh(), and execute() are awaited because they may talk to the database. This distinction helps you read async ORM code accurately.

Example 3: Transactional Transfer

A money transfer must update two rows as one unit. The endpoint validates the amount before touching the database, then opens an explicit transaction. with_for_update() asks the database to lock selected rows until the transaction ends, preventing two concurrent transfers from spending the same balance. Success returns {"status":"accepted"}. Insufficient funds returns 409 and no balances are changed.

from fastapi import Depends, FastAPI, HTTPException
from pydantic import BaseModel
from sqlalchemy import Integer, select
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column

app = FastAPI()

class Base(DeclarativeBase):
    pass

class Account(Base):
    __tablename__ = "accounts"
    id: Mapped[int] = mapped_column(Integer, primary_key=True)
    balance_cents: Mapped[int] = mapped_column(Integer, nullable=False)

class TransferRequest(BaseModel):
    from_account_id: int
    to_account_id: int
    amount_cents: int

async def get_session() -> AsyncSession:
    raise NotImplementedError

@app.post("/transfers", status_code=202)
async def transfer_money(
    payload: TransferRequest,
    session: AsyncSession = Depends(get_session),
) -> dict[str, str]:
    if payload.amount_cents <= 0:
        raise HTTPException(status_code=400, detail="Amount must be positive")

    async with session.begin():
        result = await session.execute(
            select(Account)
            .where(Account.id.in_([payload.from_account_id, payload.to_account_id]))
            .with_for_update()
        )
        accounts = {account.id: account for account in result.scalars()}
        source = accounts.get(payload.from_account_id)
        target = accounts.get(payload.to_account_id)
        if source is None or target is None:
            raise HTTPException(status_code=404, detail="Account not found")
        if source.balance_cents < payload.amount_cents:
            raise HTTPException(status_code=409, detail="Insufficient funds")
        source.balance_cents -= payload.amount_cents
        target.balance_cents += payload.amount_cents

    return {"status": "accepted"}

Row locks protect correctness but reduce concurrency for the locked accounts. The transaction should contain only the database work needed for the invariant. Do not send emails, call payment providers, or wait on unrelated APIs while holding the locks.

Example 4: Async Test Client and Test Database

Testing async database endpoints should exercise the async path. This example overrides the session dependency, creates tables, sends a request through an ASGI transport, asserts the response, then drops the tables. In a real suite, table setup normally lives in fixtures, but the ordering here shows the lifecycle clearly.

from httpx import ASGITransport, AsyncClient
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine

from app.main import Base, app, get_session

TEST_DATABASE_URL = "sqlite+aiosqlite:///./test.db"
engine = create_async_engine(TEST_DATABASE_URL)
TestingSession = async_sessionmaker(engine, expire_on_commit=False)

async def override_session():
    async with TestingSession() as session:
        yield session

async def test_create_note() -> None:
    async with engine.begin() as connection:
        await connection.run_sync(Base.metadata.create_all)
    app.dependency_overrides[get_session] = override_session

    transport = ASGITransport(app=app)
    async with AsyncClient(transport=transport, base_url="http://test") as client:
        response = await client.post("/notes", json={"title": "async path"})

    assert response.status_code == 201
    assert response.json()["title"] == "async path"

    app.dependency_overrides.clear()
    async with engine.begin() as connection:
        await connection.run_sync(Base.metadata.drop_all)

The expected behavior is a passing test with status 201 and a returned title of async path. If the endpoint accidentally uses the production dependency or a synchronous client, this test design tends to reveal it early.

Design Choices and Trade-offs

Choose a direct async driver when the query surface is small, performance sensitivity is high, and the team is comfortable writing SQL. Choose SQLAlchemy async when you need ORM identity handling, relationship mapping, reusable query construction, and migration integration. The async ORM is not a license to ignore SQL shape; inefficient joins, missing indexes, and unbounded result sets still hurt.

Pool size should reflect application concurrency and database limits. A larger pool can reduce checkout waits, but it can overload the database with more active queries. A smaller pool protects the database but increases latency during bursts. Measure checkout wait time and query latency separately before tuning.

Avoid sharing sessions. A session has mutable unit-of-work state, identity maps, and transaction context. Sharing it between requests risks cross-request data leakage, surprising flushes, and transaction conflicts. Share the engine or pool; create sessions for a specific request, job, or unit of work.

Failure Modes and Troubleshooting

Symptom: all endpoints become sluggish when one database endpoint runs. Likely cause: blocking database calls are running inside async def. Diagnose: search for synchronous driver imports and calls without await, then reproduce with concurrent requests to a cheap endpoint and the slow endpoint. Correct: switch to async driver and ORM APIs, or keep synchronous database code inside regular def routes where FastAPI can use a threadpool.

Symptom: requests hang and fail under load although database CPU is low. Likely cause: pool exhaustion from long transactions or sessions not closing. Diagnose: log pool checkout wait time, inspect active database sessions, and check dependency cleanup. Correct: return sessions promptly, reduce work inside transactions, fix leaks, and tune pool size only after leaks are removed.

Symptom: MissingGreenlet or lazy-load errors appear during response serialization. Likely cause: ORM relationships or expired attributes are being loaded after the async context where SQLAlchemy can perform awaited I/O. Diagnose: inspect response models and relationship access after commit. Correct: eager-load required relationships, refresh needed attributes explicitly, use expire_on_commit=False when appropriate, and serialize inside the session scope.

Symptom: partial changes appear after an endpoint fails. Likely cause: commits occur in multiple places. Diagnose: trace every commit() call and write a test that forces the second operation to fail. Correct: group related writes inside one async with session.begin() block and let exceptions trigger rollback.

Security, Performance, and Reliability

Parameterized queries still matter. Never build SQL by concatenating request strings. ORM filters and driver bind parameters keep user values separate from SQL text. Database credentials should be scoped to the service’s required tables and operations, and migration credentials should not be reused by the web application.

Performance depends on query plans, indexes, result sizes, pool waits, and transaction duration. Async code improves scheduling while waiting; it does not reduce the cost of a bad query. Reliability improves when every database operation has a bounded timeout, every transaction has a small purpose, and shutdown closes pools.

Hands-on Lab

Prerequisites: a FastAPI project, an async database driver for your database, SQLAlchemy async if using the ORM path, and access to a disposable local database. For SQLite experiments, use aiosqlite; for PostgreSQL, use an async dialect such as asyncpg.

Create an async engine or driver pool at application startup, keeping credentials in environment configuration.
Add a request-scoped session dependency using async with and yield.
Create a small notes table with an integer primary key and non-null title.
Add POST /notes to insert one row, commit, refresh, and return id and title.
Add GET /notes/{id} to await a select and return 404 for missing rows.
Run concurrent requests: one creating a note and one calling a cheap health endpoint. The health endpoint should still respond while the database request waits on I/O.
Add a test dependency override that points to a disposable test database and verifies the 201 response.

Verification: confirm that every database network operation is awaited, known input returns deterministic JSON, missing rows return 404, and connections return to the pool after requests finish. For PostgreSQL, active sessions before and after a short test run should not grow.

Cleanup: drop test tables or remove the disposable database file, clear FastAPI dependency overrides after tests, and close the engine or pool during shutdown. If a migration was created for the lab, roll it back in the disposable environment.

Assessment Exercises

Given an async def route that calls a synchronous database client, explain how it affects the event loop and how you would prove the problem with a concurrency test.
Design a session dependency for a background job and explain how its lifetime differs from a request-scoped dependency.
A service has high pool checkout waits but low query latency. List two likely causes and the first diagnostic measurement you would add.
Rewrite a two-step write operation so both changes commit or roll back together, and explain where validation that does not require the database belongs.
Choose between direct asyncpg and SQLAlchemy async for a reporting endpoint, naming one benefit and one cost.

Summary

Async database access in FastAPI works when the whole database path cooperates with asyncio: ASGI schedules the coroutine, the async driver awaits socket I/O, the pool controls connection concurrency, and the session or transaction scope is closed reliably. The main decisions are driver versus ORM, pool size, request-scoped session lifetime, and explicit transaction boundaries. Treat async as a concurrency tool for I/O waits, not a fix for poor SQL or unclear transactions.