Async SQLAlchemy

SQLAlchemy 2.0 has first-class asyncio support via AsyncEngine and AsyncSession. Same model, just await everywhere. The catch: lazy loading doesn’t work transparently — you must opt in to async-safe loading strategies.

Setup

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

engine = create_async_engine(
    "postgresql+asyncpg://user:pw@host/db",
    pool_size=10,
    max_overflow=20,
    pool_pre_ping=True,
    echo=False,
)

AsyncSessionLocal = async_sessionmaker(
    bind=engine,
    expire_on_commit=False,            # almost always what you want in async
    class_=AsyncSession,
)

Note the URL scheme: postgresql+asyncpg://... (not postgresql://...). The driver is asyncpg for Postgres, aiomysql for MySQL, aiosqlite for SQLite.

Basic usage

async with AsyncSessionLocal() as session:
    async with session.begin():
        session.add(User(name="alice"))
    # commits at end of inner block

    result = await session.execute(select(User))
    users = result.scalars().all()

Patterns:

• await session.execute(stmt) for queries.
• await session.commit() / await session.rollback() for transactions.
• await session.flush(), await session.refresh(obj), await session.delete(obj).
• session.add(obj) is sync (no await — it doesn’t hit the DB).

What doesn’t change

The query API is identical: select(), where(), join(), all the same. Only the I/O methods become async.

What does change — relationship loading

The big one: lazy loading is forbidden in async. The synchronous default behavior of “fetch on attribute access” can’t work — you’d need an event loop to await, but attribute access isn’t async.

async with AsyncSessionLocal() as session:
    result = await session.execute(select(User))
    user = result.scalars().first()
    print(user.posts)     # MissingGreenlet / sa.exc.InvalidRequestError

You must:

1. Eagerly load relationships via .options(selectinload(...)) or joinedload(...).
2. Or use await session.refresh(user, attribute_names=["posts"]) to load on demand.
3. Or use await session.scalars(...) with selectinload set up.

result = await session.execute(
    select(User).options(selectinload(User.posts))
)
user = result.scalars().first()
print(user.posts)     # already loaded, OK

Or via lazy=“raise” (configured on the relationship) you’d get an explicit error at definition time pushing toward eager.

await session.scalars(...)

Mirror of sync session.scalars(...):

result = await session.scalars(select(User).where(User.is_active))
users = result.all()

Note: scalars() returns a Result-like object you don’t await; the await is on session.scalars() itself.

expire_on_commit=False is the async default

In sync, after commit() all attributes are expired (re-fetched on next access). In async, accessing an expired attribute would trigger a lazy load, which is forbidden. So async setups almost always set:

async_sessionmaker(bind=engine, expire_on_commit=False)

Tradeoff: instances retain their pre-commit values; if another transaction updated the row, you have stale data until you await session.refresh(...).

AsyncSession in FastAPI

async def get_async_session() -> AsyncIterator[AsyncSession]:
    async with AsyncSessionLocal() as session:
        yield session

@app.get("/users/{uid}")
async def get_user(uid: int, session: AsyncSession = Depends(get_async_session)):
    user = await session.get(User, uid)
    return user

One async session per request. Same model as sync; just await everywhere.

run_sync — escape hatch

Some operations don’t have async wrappers (legacy hooks, Base.metadata.create_all, custom code). Use engine.begin() + run_sync:

async with engine.begin() as conn:
    await conn.run_sync(Base.metadata.create_all)

run_sync schedules the synchronous function on a thread, awaiting its completion. Don’t use for the hot path — it’s a workaround.

Async with SQLite

engine = create_async_engine("sqlite+aiosqlite:///./app.db")

Works, but SQLite’s concurrency model (one writer, multiple readers) limits async benefit. Use Postgres for real async workloads.

Performance: same pool, different driver

Underneath, asyncpg is one of the fastest Postgres drivers in any language. SQLAlchemy’s async on top adds some overhead but is usually within 2x of raw asyncpg. For tight loops over millions of rows, drop to Core (session.execute(insert(...).values(...))) or use asyncpg directly.

Pitfalls specific to async

Sharing AsyncSession across tasks

Like sync sessions, async sessions are not safe to share across concurrent tasks. Each task should get its own session.

# WRONG
async def handler(session):
    await asyncio.gather(
        do_thing_1(session),
        do_thing_2(session),
    )

# RIGHT
async def handler():
    async with AsyncSessionLocal() as s1, AsyncSessionLocal() as s2:
        await asyncio.gather(
            do_thing_1(s1),
            do_thing_2(s2),
        )

asyncpg and prepared statement caching with pgbouncer

asyncpg caches prepared statements per connection. With pgbouncer in transaction pooling mode, the connection changes per transaction, breaking the cache. Symptoms: prepared statement "__asyncpg_stmt_1__" does not exist.

Fix:

engine = create_async_engine(
    "postgresql+asyncpg://...",
    connect_args={"statement_cache_size": 0, "prepared_statement_cache_size": 0},
)

Or use a small SQLAlchemy pool (so connections live long enough to cache) and skip pgbouncer.

await not awaited

await session.execute(stmt)       # OK
session.execute(stmt)              # WRONG — coroutine warning, no SQL runs

Linters (ruff, pyright) catch this. Make sure they’re enabled.

Sync code inside async handlers

async def handler():
    # blocks the event loop:
    sync_call_to_external_api()
    # OK:
    await async_call()

The whole point of async is non-blocking I/O. Any sync blocking call (file I/O, requests library, slow Python computation) halts the event loop. Use run_in_executor or wrap in asyncio.to_thread:

import asyncio
result = await asyncio.to_thread(sync_call)

When async pays off

• High-concurrency I/O-bound services (lots of slow downstream API calls, slow DB queries).
• Streaming responses (SSE, WebSockets — see ../../12_protocols/).
• Fanning out parallel DB queries (asyncio.gather over multiple sessions).

When sync is fine:

• Most CRUD APIs on Postgres in the same datacenter.
• Apps where the DB is the bottleneck — async doesn’t make Postgres faster.
• Batch / ETL — async adds complexity for no win.

Default rule: start sync. Move to async if you’re hitting concurrency limits and downstream calls dominate.

Common interview confusions

• “Async SQLAlchemy is just sync with await.” — mostly. The catch is relationship loading: lazy loading is forbidden; you must explicitly eager-load.
• “You can share an AsyncSession across asyncio.gather tasks.” — no, not thread-safe nor task-safe. Each task needs its own session.
• “expire_on_commit=True works the same in async.” — leads to lazy load on attribute access after commit → MissingGreenlet error. Almost always set expire_on_commit=False in async sessionmaker.

Interview angle

• “How does async SQLAlchemy differ from sync?” — same API for queries, but execute/commit/flush/refresh are coroutines. Crucially: lazy loading of relationships is forbidden — you must use selectinload/joinedload explicitly.
• “Why is expire_on_commit=False standard in async setups?” — expiring attributes on commit causes next access to lazy load, which can’t work in async. Disabling expire avoids the error and gives consistent post-commit state.
• “How do you handle a forbidden lazy load in async?” — eager load with .options(selectinload(...)), or explicitly await session.refresh(obj, attribute_names=[...]).
• “What’s the URL format for async engines?” — postgresql+asyncpg://..., mysql+aiomysql://..., sqlite+aiosqlite://.... Driver name in the +driver part.
• “Why might prepared statement does not exist errors hit when using asyncpg + pgbouncer?” — pgbouncer transaction pooling reassigns DB connections per transaction; asyncpg’s per-connection prepared statement cache becomes stale. Disable with statement_cache_size=0.
• “When is async worth it for SQLAlchemy?” — high-concurrency I/O-bound services. For typical CRUD APIs, sync is simpler and the DB is usually the bottleneck anyway.