Async SQLAlchemy in Python: Engine and AsyncSession

SQLAlchemy asyncio support adapts Core and ORM to asynchronous drivers. A session remains a stateful unit of work, so every concurrent task needs its own AsyncSession.

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

engine = create_async_engine("postgresql+asyncpg://user:pass@db/app")
Session = async_sessionmaker(engine, expire_on_commit=False)


async def fetch_user(user_id: int):
    async with Session() as session:
        async with session.begin():
            return await session.scalar(
                select(User).where(User.id == user_id)
            )

Never commit credentials; obtain the URL from the deployment's secure configuration. session.begin() defines commit and rollback boundaries. Dispose of the engine during shutdown with await engine.dispose().

Avoid lazy loading that triggers implicit I/O outside an explicit await. Plan relationship loading and never call blocking database code in the event loop. The SQLAlchemy guide covers ORM basics, while Alembic covers migrations.

The official SQLAlchemy asyncio documentation, accessed July 22, 2026, covers drivers, sessions, and concurrency. Choose async when the entire request path can preserve non-blocking I/O.

One session per request or task

Create the engine once at startup and share async_sessionmaker. Open a fresh AsyncSession for each request, background job, or concurrent task. Closing the session after the unit of work finishes is part of the contract: it returns the connection to the pool and clears identity-map state.

Do not store a session on a long-lived service object and reuse it across awaits from different tasks. Concurrent use of one session can corrupt the transaction or raise errors that are hard to diagnose. Pass the session as a dependency, or open it at the edge of the use case and inject repositories that receive that session.

expire_on_commit=False is common in web apps so objects remain usable after commit without an immediate refresh. If you need fresh database state, call await session.refresh(obj) or run a new query. Prefer explicit refresh over accidental lazy loads.

Transactions and commit boundaries

async with session.begin() starts a transaction and commits on success or rolls back on exception. Nested work should use the same session and the same transaction unless you intentionally need a separate connection. Avoid mixing session.commit() calls scattered through domain code with outer context managers that also commit.

When a use case spans several writes that must succeed together, keep them inside one begin() block. If a later step fails, the earlier inserts must roll back. For read-only queries, a short session without an explicit write transaction is enough, but still close the session promptly.

Handle integrity errors at the boundary and map them to domain outcomes such as conflict or validation failure. Do not leak driver-specific exception text to API clients.

Load relationships without implicit I/O

Async SQLAlchemy discourages lazy loading because attribute access cannot safely perform I/O. Load what you need with selectinload, joinedload, or explicit queries before leaving the session:

from sqlalchemy.orm import selectinload


async def fetch_order(session, order_id: int):
    stmt = (
        select(Order)
        .where(Order.id == order_id)
        .options(selectinload(Order.items))
    )
    return await session.scalar(stmt)

Decide the loading strategy per use case. Over-fetching large graphs wastes bandwidth; under-fetching forces extra queries or broken access after the session closes. Convert ORM objects to schemas or DTOs before returning from the request handler when the response outlives the session.

Engine, pool, and driver choices

create_async_engine needs a driver that matches the dialect, such as asyncpg for PostgreSQL or aiosqlite for local SQLite tests. Pool size, overflow, and recycle settings still matter under async load: too many concurrent checkouts wait or fail, while an oversized pool can overwhelm the database.

Set statement and connection timeouts according to the driver and deployment. A hung query that never times out can exhaust the pool even if the event loop stays responsive. Measure pool checkout wait and query duration separately.

Do not wrap a synchronous Session or blocking DB-API call inside async def and call that from the event loop. If part of the stack is still synchronous, run it in a worker thread or keep that path fully sync.

Testing async repositories

Use the same async driver you care about for integration tests, or a disposable database container. An in-memory fake repository is fine for domain rules, but it cannot prove SQL, constraints, or transaction behavior.

import pytest


@pytest.fixture
async def session(engine):
    async with engine.connect() as connection:
        transaction = await connection.begin()
        Session = async_sessionmaker(bind=connection, expire_on_commit=False)
        async with Session() as session:
            yield session
        await transaction.rollback()

Confirm that your fixture's connection and transaction boundaries match how the application opens sessions. Code that creates a second engine or commits on another connection can escape the outer rollback. Prefer unique schemas or recreated databases when isolation must be stronger.

Operational checklist

Dispose the engine on shutdown. Keep credentials out of source control and logs. Prefer explicit loading over lazy access. Bound concurrency so tasks do not check out more connections than the pool can serve. Cover success, integrity conflicts, empty results, and cleanup after cancellation.

Async SQLAlchemy helps when the rest of the request path is already asynchronous. The payoff is cooperative I/O, not magic query speed. Clear session ownership and explicit transactions matter more than converting every def to async def.