SQLAlchemy Async Session in FastAPI: Patterns and Pitfalls

What is an AsyncSession and how does it fit into a FastAPI app?

An AsyncSession gives you the SQLAlchemy ORM (identity map, unit of work, relationship handling) on top of an async driver such as asyncpg. A FastAPI app uses three objects, each with a different lifetime:

• One AsyncEngine per process. Create it with create_async_engine inside FastAPI's lifespan handler and call await engine.dispose() when the handler exits. The engine owns the connection pool. Each pooled connection is an asyncpg connection bound to the event loop that opened it. Creating the engine at import time makes it easy to bind the pool to the wrong loop, especially under test runners or multiple workers.
• One async_sessionmaker. This is a configured factory and holds no state, so it is safe to share.
• One AsyncSession per request or unit of work. A session tracks loaded objects and checks out at most one connection at a time. It is not safe to use from concurrent tasks.

The engine is shared because pooled connections are expensive to open and cheap to reuse. A session is never shared because it is mutable, per-transaction state.

Internally, AsyncSession wraps a regular synchronous Session. It uses greenlets to run the sync ORM code and exposes awaitable methods at the points where database I/O happens. Anything that would trigger I/O outside those awaits, such as an implicit lazy load, fails instead of blocking the event loop.

This setup is for teams already running an async stack whose endpoints spend most of their time waiting on the database and other services. Most of the bugs it produces come from one mistake: treating AsyncSession like a thread-safe sync Session that lazily loads on attribute access. It is neither.

Copyfrom collections.abc import AsyncIterator
from contextlib import asynccontextmanager

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

DATABASE_URL = "postgresql+asyncpg://app_user:change-me@db.internal:5432/app_db"


@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncIterator[None]:
    # One engine per process, created on the running loop so the asyncpg pool binds to it.
    engine = create_async_engine(DATABASE_URL, pool_size=10, pool_pre_ping=True)
    # Stateless factory: safe to share across all requests.
    # expire_on_commit=False keeps attributes readable after commit without an implicit reload.
    app.state.sessionmaker = async_sessionmaker(engine, expire_on_commit=False)
    yield
    # Close pooled connections on the same loop that opened them.
    await engine.dispose()


app = FastAPI(lifespan=lifespan)


async def get_session(request: Request) -> AsyncIterator[AsyncSession]:
    # One AsyncSession per request. Never store it globally or share it between tasks.
    async with request.app.state.sessionmaker() as session:
        yield session


@app.get("/health/db")
async def db_health(session: AsyncSession = Depends(get_session)) -> dict:
    # Database I/O happens only at awaited calls like execute().
    result = await session.execute(text("SELECT 1"))
    return {"db": result.scalar_one()}

A minimal FastAPI app that creates one AsyncEngine in the lifespan handler, shares a single async_sessionmaker, and gives each request its own AsyncSession through a dependency.

How should you scope the session and transaction per request?

The dependency should own the session's lifetime but not its transaction. Open it with async with sessionmaker() as session, yield it, and let the context manager close it. Closing rolls back anything uncommitted and returns the connection to the pool. The commit belongs where the business operation ends, so the code that knows the work is complete is the code that makes it durable.

Use session.begin() blocks in the service layer:

• On success: the block exits and commits, and then the endpoint returns.
• On an HTTPException raised inside the block: the block rolls back.
• On an HTTPException raised after the block: the earlier commit stands.
• On an unexpected error: the block rolls back, the dependency's context manager closes the session, and the error propagates to FastAPI's handlers.

Commit-after-yield looks tidier, but the endpoint has already chosen a status code and body by the time teardown runs. A constraint violation at that point is reported late or lost, depending on how your FastAPI version orders teardown and response sending.

There is also a resource cost. A session holds a pooled connection from its first query until the transaction ends. If the transaction only ends at teardown, an endpoint that runs one query and then waits seconds on an external API keeps that connection for the whole wait. Under load, those idle-in-transaction connections exhaust the pool. For slow endpoints that do little database work, open a short session inside the service, finish the unit of work, and release the connection before the slow call.

Copyfrom typing import Annotated, AsyncIterator
from fastapi import Depends, FastAPI, HTTPException
from sqlalchemy import text
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine

engine = create_async_engine("postgresql+asyncpg://app:change-me@db.internal:5432/shop")
SessionLocal = async_sessionmaker(engine, expire_on_commit=False)

async def get_session() -> AsyncIterator[AsyncSession]:
    # Owns the session's lifetime only. No commit here: exiting the context
    # closes the session, rolling back anything uncommitted and returning the connection.
    async with SessionLocal() as session:
        yield session

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

async def reserve_stock(session: AsyncSession, sku: str, qty: int) -> int:
    # The unit of work. begin() must open the transaction before any query runs
    # on this session, or it raises because autobegin already started one.
    async with session.begin():
        result = await session.execute(
            text("UPDATE stock SET qty = qty - :q WHERE sku = :s AND qty >= :q RETURNING qty"),
            {"q": qty, "s": sku},
        )
        remaining = result.scalar_one_or_none()
        if remaining is None:
            # Raised inside the block, so the UPDATE is rolled back.
            raise HTTPException(status_code=409, detail="insufficient stock")
    # Clean exit committed. Commit errors surface here, before any response exists.
    return remaining

@app.post("/stock/{sku}/reserve")
async def reserve(sku: str, qty: int, session: SessionDep) -> dict:
    remaining = await reserve_stock(session, sku, qty)
    # An HTTPException raised here would not undo the committed reservation.
    return {"sku": sku, "remaining": remaining}

A yield dependency that only opens and closes the AsyncSession, with the service layer's async with session.begin() block deciding where commit and rollback happen, including an HTTPException raised inside the block.

Why does accessing attributes raise MissingGreenlet, and how do you load data explicitly?

SQLAlchemy's async layer wraps a synchronous core. Every database call has to pass through an explicit await so the greenlet bridge can hand control to the event loop. Attribute access is a plain Python __getattr__ with no await point. When touching an attribute needs a query, such as an expired column after commit() or an unloaded lazy relationship, SQLAlchemy has no way to yield to the loop. It raises MissingGreenlet instead of blocking.

Set expire_on_commit=False on the session factory. Without it, every commit expires all loaded state, and the next attribute read during response serialization fails. The cost is that objects keep their pre-commit values. Server defaults, sequence-generated columns other than the primary key, onupdate server expressions and values changed by triggers stay stale until you refresh them. Use await session.refresh(obj, ["updated_at"]) to fetch only the columns you need.

For relationships, choose the loading strategy in each query rather than on the model. One endpoint needs the children, and another serving the same model may not.

Copyfrom datetime import datetime
from sqlalchemy import ForeignKey, func, select
from sqlalchemy.ext.asyncio import AsyncAttrs, async_sessionmaker, create_async_engine
from sqlalchemy.orm import DeclarativeBase, Mapped, joinedload, mapped_column, relationship, selectinload

class Base(AsyncAttrs, DeclarativeBase):  # AsyncAttrs adds obj.awaitable_attrs
    pass

class Author(Base):
    __tablename__ = "author"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str]
    updated_at: Mapped[datetime] = mapped_column(server_default=func.now(), onupdate=func.now())
    # Fail loudly: any access without an explicit loader option raises
    books: Mapped[list["Book"]] = relationship(back_populates="author", lazy="raise")

class Book(Base):
    __tablename__ = "book"
    id: Mapped[int] = mapped_column(primary_key=True)
    author_id: Mapped[int] = mapped_column(ForeignKey("author.id"))
    # Default lazy load, so awaitable_attrs can fetch it on demand
    author: Mapped[Author] = relationship(back_populates="books")

engine = create_async_engine("postgresql+asyncpg://app:change-me@localhost/appdb")
# Keep loaded state after commit so response serialization doesn't trigger I/O
SessionLocal = async_sessionmaker(engine, expire_on_commit=False)

async def examples() -> None:
    async with SessionLocal() as session:
        # Collection: selectinload issues one extra SELECT ... WHERE author_id IN (...)
        authors = (await session.scalars(select(Author).options(selectinload(Author.books)))).all()
        # joinedload on a collection duplicates parent rows: .unique() is required
        same = (await session.scalars(select(Author).options(joinedload(Author.books)))).unique().all()
        # Many-to-one: joinedload adds no round trip and no duplication
        books = (await session.scalars(select(Book).options(joinedload(Book.author)))).all()

        author = authors[0]
        author.name = "Renamed"
        await session.commit()
        # updated_at was set by SQL; reload just that column instead of touching it lazily
        await session.refresh(author, ["updated_at"])

        plain = await session.get(Book, 1)  # author not loaded here
        owner = await plain.awaitable_attrs.author  # one awaited query; avoid in loops (N+1)

Models with lazy="raise" and AsyncAttrs, a session factory with expire_on_commit=False, and per-query loading via selectinload, joinedload with .unique(), targeted session.refresh, and awaitable_attrs.

How do you return ORM data through Pydantic response models safely?

When a handler returns an ORM instance and lets response_model do the conversion, FastAPI serializes the object after your function has returned. Pydantic walks every field in the response schema and reads the matching attribute on the instance. That walk is synchronous. If a field maps to a relationship you did not load, SQLAlchemy tries to lazy load it from a context that cannot await, and you get MissingGreenlet, or the lazy="raise" error if you set that default. If the request's transaction has already committed and expire_on_commit is left at its default, plain column attributes are expired too and need a refresh you cannot perform there. The failure surfaces in framework code, with a traceback that points at serialization rather than at the query that should have loaded the data.

The safer rule is structural: convert to Pydantic inside the session scope, and return the schema rather than the ORM object. In the service or handler, issue the query with explicit loader options, such as selectinload for collections and joinedload for many-to-one, covering everything the schema touches. Then call model_validate on the result, using a schema configured with from_attributes=True, before the session closes.

This saves debugging time because loading requirements become visible in the query. When someone adds a nested field to a response schema, the missing load fails on the conversion line, inside your code, the first time a test exercises that endpoint. It does not fail intermittently during serialization. Each endpoint's query states exactly what that endpoint needs, so reviewers can check eager loads against the schema in one place.

Copyfrom fastapi import APIRouter, Depends, HTTPException
from pydantic import BaseModel, ConfigDict
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import joinedload, selectinload

from app.db import get_session  # yields one AsyncSession per request
from app.models import Order    # Order.customer (many-to-one), Order.items (collection)

router = APIRouter()

class ItemOut(BaseModel):
    model_config = ConfigDict(from_attributes=True)
    sku: str
    quantity: int

class CustomerOut(BaseModel):
    model_config = ConfigDict(from_attributes=True)
    id: int
    name: str

class OrderOut(BaseModel):
    model_config = ConfigDict(from_attributes=True)
    id: int
    customer: CustomerOut   # requires Order.customer to be loaded
    items: list[ItemOut]    # requires Order.items to be loaded

@router.get("/orders/{order_id}", response_model=OrderOut)
async def get_order(order_id: int, session: AsyncSession = Depends(get_session)) -> OrderOut:
    # Every relationship OrderOut reads is declared here, next to the query.
    stmt = (
        select(Order)
        .where(Order.id == order_id)
        .options(joinedload(Order.customer), selectinload(Order.items))
    )
    order = await session.scalar(stmt)
    if order is None:
        raise HTTPException(status_code=404, detail="Order not found")
    # Convert while the session is open: a missing loader fails on this line, in our code.
    return OrderOut.model_validate(order)  # return the schema, never the ORM object

A FastAPI handler that eager-loads every relationship the response schema needs and converts the ORM result to a Pydantic model with model_validate before the session closes, so serialization never triggers a lazy load.

When should you avoid sharing an AsyncSession or holding it open?

An AsyncSession is not safe for concurrent use. It holds a single connection and one identity map, and its state changes at every await. If two tasks interleave operations on the same session, for example under asyncio.gather, you get errors about operations already in progress or, worse, a corrupted unit of work. Give each concurrent unit of work its own session from the sessionmaker, and do not try to parallelize queries inside one session.

FastAPI BackgroundTasks run after the response is sent. By then the request-scoped session may already be closed, or its transaction committed. A background task should open its own session and receive only plain identifiers, not ORM instances bound to the request's session.

Pool pressure

pool_size plus max_overflow caps how many connections the engine will hold at once, and that cap limits your real database concurrency. pool_pre_ping checks a connection before handing it out, which catches connections dropped by a firewall or a database restart, at the cost of one round trip per checkout.

A session holds its connection from the first query until commit or close. If a handler streams a large response, or calls an external HTTP API while a transaction is open, that connection sits idle and other requests wait for one. Read what you need, commit, close the session, and then do the external I/O.

Async does not make database work faster. Hydrating thousands of ORM objects is CPU-bound and blocks the event loop. A small pool still queues requests, just as it does with sync code.

What breaks with asyncpg behind PgBouncer-style poolers, and what are the alternatives?

The failure starts with asyncpg's prepared statements. asyncpg prepares statements on the server and caches them per connection, and SQLAlchemy's asyncpg dialect keeps its own prepared-statement cache on top of that. Both assume that the connection you hold is always the same backend. A transaction-mode pooler breaks that assumption. Between transactions, it can route your client connection to a different server connection. A statement prepared on backend A does not exist on backend B, so you get "prepared statement does not exist" errors. You can also get name collisions when another client already used that name on the same backend. These errors are intermittent and depend on load, which makes them hard to reproduce locally.

The SQLAlchemy asyncpg dialect documentation describes how to disable both caches. Check it for the current options instead of copying parameters from old answers.

For sync-only APIs such as metadata.create_all or legacy helpers, run_sync on AsyncConnection or AsyncSession passes in a sync connection or session as an escape hatch. Another approach is a separate sync engine for migrations and scripts. It works, but it duplicates URLs, pool settings, and connect arguments, and the two configurations tend to drift apart.

If your endpoints are mostly CRUD and you have no real concurrency pressure, use plain def endpoints with a sync Session. FastAPI runs them in a threadpool, and you avoid this whole category of async pitfalls.

How do you test async SQLAlchemy sessions in FastAPI and keep development cost down?

Most of the test setup is one fixture and one dependency override. For each test, open a connection from the engine and begin a transaction on it. Then create an AsyncSession bound to that connection with join_transaction_mode="create_savepoint", so a commit() in application code releases a savepoint and does not commit the outer transaction. Register a replacement for your session dependency through app.dependency_overrides that yields this session. When the test finishes, roll back the outer transaction and close the connection. Every test then sees a clean database without truncating tables, and you are testing the real commit paths, not mocks.

Event loops cause most of the flaky failures here. An asyncpg connection belongs to the event loop that created it. Suppose you build the engine in a session-scoped fixture, but pytest-asyncio runs each test in its own function-scoped loop. The pool will hand later tests connections from a dead loop, and you'll get errors about futures attached to a different loop, or hangs at teardown. Keep the scopes consistent. You can create the engine in a fixture with the same loop scope as the tests, configure a single shared loop scope for the whole suite, or use NullPool in tests so no connection outlives its loop.

The earlier decisions all add up-front cost: one session per request, a deliberate expire_on_commit setting, lazy="raise", and converting to schemas inside the session scope. Scripts get wordier, and every relationship access needs an explicit load. In return, intermittent MissingGreenlet errors, stale reads and pool exhaustion show up as deterministic failures in tests, not in production. The main risk is erosion. When a deadline hits, someone flips lazy="raise" back to the default "just for this model", and the hidden lazy loads come back. Enforce the setting in a shared base class and have code review reject per-model overrides.

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 sqlalchemy async session fastapi, 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 route's URL or method automatically notifies the services that consume it — no hunting through repos for callers.

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.
• Consumers learn about a breaking change at review time rather than from production errors.
• 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.