FastAPI Async Database Access: SQLAlchemy Async vs asyncpg

How SQLAlchemy async and asyncpg relate

asyncpg is a PostgreSQL driver for asyncio. Rather than wrapping libpq, it implements the PostgreSQL wire protocol itself and uses the binary format for data exchange, so rows decode into Python types with little overhead. It ships its own connection pool and exposes PostgreSQL-specific features such as prepared statements, COPY, and custom type codecs.

SQLAlchemy's async support does not contain a driver. AsyncEngine and AsyncSession are asyncio-facing wrappers around SQLAlchemy's synchronous Engine and Session. SQLAlchemy runs its internals inside greenlets, which lets that existing synchronous code call an async driver without being rewritten. Point the engine at a postgresql+asyncpg URL and asyncpg does the network work underneath. SQLAlchemy also manages its own connection pool, so asyncpg's built-in pool is not involved in that setup.

Because both options run on asyncpg, choosing between them is not a choice of driver. The real question is how much abstraction to put on top of asyncpg:

• Raw asyncpg: hand-written SQL and the least overhead.
• SQLAlchemy Core: composable SQL expressions and dialect handling, with no identity map.
• ORM: mapped objects, unit-of-work, and relationship loading.

Each step up adds per-row Python work and new ways for things to fail.

None of these layers makes the database faster. Async lets the event loop serve other requests while one request waits on a query. That query still holds a pooled connection and still uses database CPU and I/O. Once the pool is exhausted or PostgreSQL is saturated, more coroutines only make the queue longer.

Copy+-------------------------------------------------------------------+
| FastAPI on the asyncio event loop: many request coroutines        |
+-------------------------------------------------------------------+
        v                                          v
+-------------------------+         +-------------------------------+
| Raw asyncpg             |         | ORM: AsyncSession             |
| hand-written SQL        |         | mapped objects, unit of work  |
| least overhead          |         | relationship loading          |
+-------------------------+         +-------------------------------+
        |                                          v
        |                           +-------------------------------+
        |                           | Core: AsyncEngine             |
        |                           | SQL expressions, dialect      |
        |                           | greenlets run sync Engine     |
        |                           +-------------------------------+
        v                                          v
+-------------------------+         +-------------------------------+
| asyncpg pool            |         | SQLAlchemy pool               |
+-------------------------+         +-------------------------------+
        v                                          v
+-------------------------------------------------------------------+
| asyncpg: own wire protocol, binary format (used by both)          |
+-------------------------------------------------------------------+
        v   each query holds one pooled connection until it returns
+-------------------------------------------------------------------+
| PostgreSQL: CPU and I/O set the throughput ceiling                |
| more coroutines past this point only lengthen the queue           |
+-------------------------------------------------------------------+

Raw asyncpg and SQLAlchemy's async stack (ORM over Core over AsyncEngine) both run on the asyncpg driver and use separate pools, and every path ends at the same PostgreSQL capacity limit.

Wiring the engine or pool into FastAPI

…queue for a connection, so the wiring has to make pool ownership explicit.

Both options follow the same shape: one long-lived pool per process, and one short-lived borrower per request. Create the pool inside FastAPI's lifespan handler, not at import time. Import-time creation runs before the event loop the server uses exists, and it opens connections in every process that imports the module, including test collectors and migration scripts. In the lifespan handler, build an AsyncEngine with create_async_engine plus an async_sessionmaker, or call asyncpg.create_pool. Store the result on app.state. After the handler's yield, call await engine.dispose() or await pool.close() so connections close cleanly on shutdown instead of being dropped by the server.

Hand out per-request resources through a dependency that uses yield. FastAPI runs the code after the yield whether the endpoint returns or raises. That is where the connection goes back to the pool.

• SQLAlchemy: open the session with async with sessionmaker() as session: and yield it. Exiting the context closes the session and returns its connection, rolling back any uncommitted transaction. Many teams also set expire_on_commit=False so attributes stay readable after commit without triggering a reload.
• asyncpg: use async with pool.acquire() as conn: and yield conn. Exiting the context releases the connection and resets its session state. The transaction is not managed for you, so wrap writes in conn.transaction() explicitly.

Never acquire inside the dependency without a context manager. If an exception skips the release, that connection leaks, and the pool shrinks for the rest of the process's life.

Copyfrom contextlib import asynccontextmanager

import asyncpg
from fastapi import Depends, FastAPI, Request
from sqlalchemy import text
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine


@asynccontextmanager
async def lifespan(app: FastAPI):
    # Both pools in one app only for comparison; a real service picks one.
    # Created here, after the server's event loop exists, never at import time.
    engine = create_async_engine("postgresql+asyncpg://app:change-me@db:5432/appdb", pool_size=10)
    app.state.engine = engine
    app.state.sessionmaker = async_sessionmaker(engine, expire_on_commit=False)
    app.state.pool = await asyncpg.create_pool("postgresql://app:change-me@db:5432/appdb", min_size=2, max_size=10)
    yield
    # Shutdown: close connections cleanly instead of letting the server drop them.
    await app.state.pool.close()
    await engine.dispose()

app = FastAPI(lifespan=lifespan)

async def get_session(request: Request):
    # Leaving the block closes the session, returns its connection, and rolls back uncommitted work.
    async with request.app.state.sessionmaker() as session:
        yield session

async def get_conn(request: Request):
    # Leaving the block releases the connection to the pool even if the endpoint raised.
    async with request.app.state.pool.acquire() as conn:
        yield conn

@app.get("/orm/orders/count")
async def count_orders(session: AsyncSession = Depends(get_session)):
    result = await session.execute(text("SELECT count(*) FROM orders"))
    return {"count": result.scalar_one()}

@app.post("/raw/orders")
async def create_order(conn: asyncpg.Connection = Depends(get_conn)):
    async with conn.transaction():  # asyncpg leaves transactions to you
        order_id = await conn.fetchval("INSERT INTO orders (status) VALUES ('new') RETURNING id")
    return {"id": order_id}

A lifespan handler creates one SQLAlchemy AsyncEngine and one asyncpg pool per process and closes them on shutdown. Matching yield dependencies hand out an AsyncSession or a pooled connection per request and always release it, even when the request fails.

Session scope, concurrency, and the lazy-loading trap

Treat an AsyncSession the same way as a pooled connection: one per request or unit of work, created by a dependency and closed when it exits. A session is a stateful object. It holds one connection, one transaction, and an identity map, and it is not safe for concurrent use. If you pass the same session into several coroutines under asyncio.gather, they interleave on a single connection, and SQLAlchemy raises an error about concurrent operations or an illegal state change. To run queries in parallel, give each task its own session from the async_sessionmaker. Each of those sessions checks out its own connection, so a request now holds several pool slots at once. Account for that when you size the pool.

Lazy loading is the most common failure when migrating sync code. Touching an unloaded relationship in sync SQLAlchemy quietly issues a query. Under asyncio there is no awaitable point at that moment, so you get a MissingGreenlet error. Once the session has closed, you get a detached-instance error instead. The fix is to load what you need explicitly with selectinload or joinedload, and to set expire_on_commit=False so attributes don't expire after commit and trigger a hidden refresh. Setting lazy="raise" on relationships turns any accidental lazy load into an immediate error rather than a production surprise.

Expect this refactor to touch every query path that relies on implicit relationship access. Budget for that work.

Finally, don't hold a transaction open while you call a slow external API. Commit first, make the call, then open a new unit of work. A transaction left waiting on a network call keeps its connection checked out and can keep row locks held for the whole wait.

Copyimport asyncio
from fastapi import Depends, FastAPI
from sqlalchemy import func, select
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
from sqlalchemy.orm import selectinload
from app.models import Order, User  # User.orders = relationship(..., lazy="raise")

engine = create_async_engine("postgresql+asyncpg://app:change-me@db:5432/shop", pool_size=10)
# expire_on_commit=False: loaded attributes stay readable after commit (no hidden refresh)
SessionLocal = async_sessionmaker(engine, expire_on_commit=False)
app = FastAPI()

async def get_session():
    async with SessionLocal() as session:  # one session per request, closed on exit
        yield session

async def count(model) -> int:
    async with SessionLocal() as s:  # each concurrent task gets its OWN session/connection
        return await s.scalar(select(func.count()).select_from(model))

@app.get("/stats")
async def stats():
    # Two sessions run in parallel, so this request holds two pool slots
    users, orders = await asyncio.gather(count(User), count(Order))
    return {"users": users, "orders": orders}

@app.get("/users/{user_id}")
async def read_user(user_id: int, session: AsyncSession = Depends(get_session)):
    # Eager-load explicitly; an implicit lazy load would raise instead of querying
    stmt = select(User).options(selectinload(User.orders)).where(User.id == user_id)
    user = await session.scalar(stmt)
    return {"id": user.id, "order_ids": [o.id for o in user.orders]}

@app.post("/users/{user_id}/notify")
async def notify(user_id: int, session: AsyncSession = Depends(get_session)):
    user = await session.get(User, user_id)
    user.notified = True
    await session.commit()  # commit BEFORE the slow call: releases locks, ends the transaction
    await send_email(user.email)  # placeholder for a slow external API call
    return {"ok": True}

A FastAPI app using one AsyncSession per request. It gives each asyncio.gather task its own session, eager-loads relationships with selectinload under expire_on_commit=False, and commits before making a slow external call.

Raw asyncpg and SQLAlchemy Core as alternatives to the ORM

If the ORM's unit-of-work machinery is more than a service needs, you can drop to asyncpg directly. Queries use PostgreSQL's native positional placeholders ($1, $2) rather than %s or named parameters. Pick the method by result shape: fetch returns a list of Record objects, fetchrow returns one Record or None, fetchval returns a single scalar, and execute returns the command status string (such as INSERT 0 1). Multi-statement work goes inside async with conn.transaction():, which commits on clean exit and rolls back on exception.

What you take on

Record supports key access, so Model(**dict(record)) turns a row into a Pydantic model. You write that mapping for every query, though, and nothing catches a renamed column until runtime. Schema migrations are also yours to handle, because there is no metadata for a tool to diff. Your SQL is PostgreSQL-specific by construction. The biggest risk is dynamic SQL: optional filters and user-chosen sort columns push people toward f-strings. Placeholders bind values only, never identifiers, so column and table names must come from an allowlist.

The middle path: Core on an async engine

SQLAlchemy Core runs on the same AsyncEngine used for the ORM, through async with engine.begin() as conn: await conn.execute(...). You build queries from Table objects with select(), where(), and order_by(), so adding optional filters means composing Python expressions instead of concatenating strings, and every value is bound. The MetaData that defines those tables is what Alembic's autogenerate compares against, which gives you migrations. Results come back as plain rows, so there is no identity map, lazy loading, or session state to reason about. asyncpg still does the I/O underneath.

Copyimport asyncpg
from pydantic import BaseModel

class Order(BaseModel):
    id: int
    customer_id: int
    status: str
    total_cents: int

COLS = "id, customer_id, status, total_cents"
# Placeholders bind values only; identifiers like sort columns need an allowlist.
SORT_COLUMNS = {"id", "created_at", "total_cents"}

async def get_order(conn: asyncpg.Connection, order_id: int) -> Order | None:
    row = await conn.fetchrow(f"SELECT {COLS} FROM orders WHERE id = $1", order_id)
    return Order(**dict(row)) if row else None  # hand-written Record -> model mapping

async def list_orders(conn: asyncpg.Connection, customer_id: int,
                      status: str | None, sort: str) -> list[Order]:
    if sort not in SORT_COLUMNS:
        raise ValueError(f"unsupported sort column: {sort!r}")
    sql, args = f"SELECT {COLS} FROM orders WHERE customer_id = $1", [customer_id]
    if status is not None:
        args.append(status)
        sql += f" AND status = ${len(args)}"  # only the placeholder number is formatted in
    sql += f" ORDER BY {sort} DESC"  # safe only because sort passed the allowlist
    return [Order(**dict(r)) for r in await conn.fetch(sql, *args)]

async def place_order(pool: asyncpg.Pool, customer_id: int, total_cents: int) -> int:
    async with pool.acquire() as conn:
        async with conn.transaction():  # commits on clean exit, rolls back on exception
            order_id = await conn.fetchval(
                "INSERT INTO orders (customer_id, status, total_cents) "
                "VALUES ($1, 'pending', $2) RETURNING id", customer_id, total_cents)
            tag = await conn.execute(
                "UPDATE customers SET order_count = order_count + 1 WHERE id = $1", customer_id)
            if tag != "UPDATE 1":  # execute returns the command status string
                raise LookupError(f"customer {customer_id} not found")  # undoes the INSERT
            return order_id

Raw asyncpg in practice: $n placeholders, fetchrow/fetch/fetchval/execute, a transaction block that rolls back on error, manual Record-to-Pydantic mapping, and an allowlist guarding a dynamic ORDER BY column.

Connection pool sizing and transaction-mode poolers

A connection pool holds a fixed set of open connections that requests borrow and return. In SQLAlchemy, pool_size is the number of connections kept open, and max_overflow is how many extra connections may be opened temporarily under load and closed once they are returned. Each Uvicorn worker is a separate process with its own pool, and each pod runs its own workers, so the totals multiply.

Illustrative estimate

Assume 4 Uvicorn workers per pod, 3 pods, pool_size=10 and max_overflow=5:

4 × 3 × (10 + 5) = 180 potential connections

That total must stay below PostgreSQL's max_connections, leaving headroom for migrations, admin sessions, and monitoring. Every PostgreSQL connection is a backend process with its own memory, so raising max_connections to fit has a real cost. A rolling deploy also briefly runs old and new pods side by side, which can push the total above your steady-state number.

Making pools smaller does not remove the demand. When every connection is checked out, new requests wait in the pool's queue. If one waits longer than pool_timeout, it fails with a timeout error. Latency climbs before any errors appear. When replica counts make per-process pools impractical, put an external pooler such as PgBouncer in front of PostgreSQL.

In transaction mode, the pooler can send consecutive transactions from one client to different server connections. asyncpg caches prepared statements per connection, so a cached statement may not exist on the backend that receives the next query, and the query fails. The usual fix is to disable or limit the statement cache in both asyncpg and SQLAlchemy's asyncpg dialect. Option names and defaults change between releases, so check the exact settings in the current asyncpg and SQLAlchemy documentation.

Avoiding event-loop blocking and testing pitfalls

A synchronous driver call inside an async def endpoint does not just slow one request. It halts the event loop, so every other in-flight request on that worker stops until the call returns. The same applies to a sync Session used by mistake, or a sync library that opens its own connection. If a dependency only offers a blocking API, declare the endpoint with plain def so FastAPI runs it in a threadpool, or offload the call explicitly with asyncio.to_thread.

Async code also needs time limits. An awaited query does not block the loop, but a runaway one still holds a pooled connection indefinitely. Set PostgreSQL's statement_timeout per role or per connection so the server enforces a ceiling. Large result sets cause a quieter version of the same problem: decoding thousands of rows into ORM objects, or serialising them, is CPU work that runs on the loop. Paginate the query, or move heavy transformation to a thread or process.

Most flaky async tests have one of three causes:

• Fixture loop scope. asyncpg connections are bound to the loop that created them. A session-scoped engine combined with function-scoped loops produces "attached to a different loop" errors. Align the scopes, or use NullPool in tests.
• Per-test rollback. Wrap each test in an outer transaction that is rolled back at the end, and make the app's dependency reuse that connection.
• Shared engines across loops. Dispose the engine when its loop closes.

Alembic and background workers often end up with a second, synchronous engine configuration that drifts from the async one. To avoid this, run migrations through the async engine with connection.run_sync, and build both engines from a single settings object.

Choosing the right layer: decision guide and engineering cost

Each layer saves engineering time in one place and costs it in another. The right choice depends on where your team's hours actually go.

ORM overhead is paid by CPU on every query. Hand-written mapping is paid by engineers for every query and again for every schema change. Compare the two directly.

Illustrative estimate (assumed numbers, not a benchmark):

• Mapping cost = queries × hours per query = 60 × 0.75 h = 45 engineer-hours, plus rework whenever the schema changes.
• ORM overhead = requests/s × queries per request × extra CPU per query = 200 × 3 × 0.2 ms = 120 ms of CPU per second, roughly 12% of one core.

At these inputs, one more replica or core is cheaper than 45 hours of mapping code. The trade reverses for a service handling thousands of requests per second with five queries total.

Generic benchmarks measure someone else's queries. Profile your own hot endpoints under realistic concurrency before you rule out a layer.

How OneEnv can help

Disclosure: OneEnv is the product behind this blog.

OneEnv (oneenv.io) is a workspace-based governance platform for the things services share — environment variables, database structure, API routes and validation rules. A change to a shared value is staged on a branch, reviewed once, and then every affected service explicitly approves or rejects it before it goes live, so you can see who a change touches before it ships instead of finding out in production.

Applied to fastapi async database, this is where it helps:

Saves time

• Every service that inherits a shared value is listed in one place, so nobody searches repos and chats to find out who is affected.
• One review per branch replaces coordinating each team separately; each service still keeps its own approve/reject.
• Changing a shared variable is one edit and one review instead of editing every service's .env by hand.

Saves cost

• Fewer production incidents and rollbacks caused by a shared value, route or column changing under a consumer unannounced.
• Less duplicated configuration to maintain: one shared definition inherited by many services.
• A wrong or stale value is caught at review time by the services it affects, not by a failed deploy.
• OneEnv has a free plan (1 workspace, unlimited services and members, full governance), so trying it costs nothing.

These are qualitative benefits, not measured results — how much you save depends on how many services share the value.

Want to try it? Start at oneenv.io.