SQLAlchemy 2.0 async ORM in production

SQLAlchemy 2.0 unifies the Core SELECT/INSERT API with the ORM, replaces Query with select(), and ships first-class async support across the stack.

Two consequences for backend code:

• One API surface. session.execute(select(User).where(...)) looks identical to a Core query against the same engine. No more session.query(User) vs engine.execute(text("...")) split.
• An honest async path. AsyncEngine + AsyncSession wrap the sync ORM with awaitable operations. Pre-2.0 async stories were either ad-hoc wrappers or non-ORM. Now the async session is mainstream and the patterns are settled.

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

engine = create_async_engine("postgresql+asyncpg://...")

async def get_user(uid: int) -> User:
    async with AsyncSession(engine) as session:
        result = await session.execute(select(User).where(User.id == uid))
        return result.scalar_one()

The driver matters: async needs an async driver (asyncpg, aiomysql, aiosqlite). The sync psycopg2 will not work behind create_async_engine — SQLAlchemy rejects the URL at construction time.

Engines, sessions, and transactions are different things with different lifetimes:

• Engine — a connection-pool manager. Process-lifetime. One per database URL. Cheap to share; expensive to construct.
• Session — a unit-of-work tracker. Request-lifetime. Holds the identity map and the pending changes. Cheap to create; dangerous to share across requests.
• Transaction — the actual DB transaction. Opened explicitly with session.begin() or implicitly on the first session.execute().

# Process startup — one engine with a pool.
engine = create_async_engine(DATABASE_URL, pool_size=20, max_overflow=10)

# Per request — one session, one transaction.
async with AsyncSession(engine) as session:
    async with session.begin():
        session.add(User(email="a@b.com"))
    # On exit success: COMMIT. On exception: ROLLBACK.

Mixing layers is the classic mistake: an engine per request kills the pool, a session held for the whole process holds locks forever and grows the identity map without bound. Anchor the engine at process start, build one session per request.

from datetime import datetime
from sqlalchemy import String, ForeignKey
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "users"
    id: Mapped[int] = mapped_column(primary_key=True)
    email: Mapped[str] = mapped_column(String(255), unique=True)
    created_at: Mapped[datetime]
    posts: Mapped[list["Post"]] = relationship(back_populates="author")

class Post(Base):
    __tablename__ = "posts"
    id: Mapped[int] = mapped_column(primary_key=True)
    author_id: Mapped[int] = mapped_column(ForeignKey("users.id"))
    title: Mapped[str]
    author: Mapped[User] = relationship(back_populates="posts")

Mapped[T] is read at class-creation time. Mapped[int] becomes a NOT NULL integer column. Mapped[int | None] allows NULL. Mapped[list[Post]] is a one-to-many relationship. The annotation drives both the schema and the static type of user.id, so the same line that defines the column also gives mypy/pyright enough to type-check the queries against it.

SQLAlchemy is not a "run a query, get a row" layer. It is a unit of work: the session tracks every loaded object and every modification, and commit() flushes the minimal set of statements that reconciles them.

async with AsyncSession(engine) as session:
    u1 = await session.get(User, 1)
    u2 = await session.get(User, 1)
    assert u1 is u2          # same Python object — identity map at work

    u1.email = "new@example.com"
    # No SQL yet. The session marks u1 dirty.

    other = User(email="x@y.com")
    session.add(other)
    # No SQL yet. The session marks `other` pending.

    await session.commit()
    # Now: BEGIN, UPDATE users WHERE id=1, INSERT INTO users, COMMIT.

The identity map means two session.get(User, 1) calls return the same Python object — convenient for relationship navigation, hazardous if you hold a session across requests (stale data, locks held, memory growth). Closing the session drops the identity map; the next session rebuilds it.

The single most expensive bug in any ORM-backed backend:

async def list_users():
    result = await session.execute(select(User))
    users = result.scalars().all()
    return [{"email": u.email, "posts": len(u.posts)} for u in users]

Reading u.posts triggers SELECT * FROM posts WHERE author_id = ? per user. 50 users → 51 queries. The fix is to tell the session to load the relationship up front:

from sqlalchemy.orm import selectinload, joinedload

# selectinload — separate SELECT ... IN (...). Best for to-many.
result = await session.execute(
    select(User).options(selectinload(User.posts))
)

# joinedload — single LEFT JOIN. Best for to-one or small to-many.
result = await session.execute(
    select(Post).options(joinedload(Post.author))
)

Detection in dev: create_async_engine(URL, echo="debug") logs every statement — one SELECT users followed by 50 SELECT posts is the visual fingerprint. In production: slow-query logs or APM (Datadog, OpenTelemetry instrumentation for SQLAlchemy) aggregate the same pattern.

The rule that catches every team migrating from sync SQLAlchemy: AsyncSession does not allow lazy loading. Accessing an unloaded relationship would issue a synchronous query inside an async context, defeating the whole point of async.

async with AsyncSession(engine) as session:
    u = await session.get(User, 1)
    print(u.posts)   # sqlalchemy.exc.MissingGreenlet: not loaded

Three correct patterns:

• Eager load up front with selectinload(User.posts) or joinedload(...) in the select().
• await session.refresh(u, ["posts"]) to pull specific relationships after the fact.
• await session.run_sync(callback) to drop into sync mode for a tightly scoped block — escape hatch, not a default.

The error message — MissingGreenlet: greenlet_spawn has not been called — is the migration-playbook smell. Default to eager loading; the up-front query is almost always cheaper than the latency hit of a forgotten relationship load.

The session.begin() block is the cleanest unit of work:

async with AsyncSession(engine) as session:
    async with session.begin():
        session.add(User(email="a@b.com"))
        await session.execute(
            update(Order).where(Order.id == oid).values(status="paid")
        )
    # Exit success → COMMIT. Exit on exception → ROLLBACK.

Nested begin_nested() opens a SAVEPOINT — useful when a sub-step is allowed to fail without aborting the outer transaction:

async with session.begin():
    session.add(parent)
    try:
        async with session.begin_nested():       # SAVEPOINT
            session.add(child_that_might_violate)
    except IntegrityError:
        pass   # SAVEPOINT rolled back; outer transaction alive

Don't call await session.commit() inside a begin() block — it is redundant and raises one of the library's more confusing exceptions. Pick one shape: either the begin() context manager, or manual commit() / rollback(). Mixing them is the source of most "transaction already begun" errors.

Three knobs that decide whether the backend stays up:

• pool_size — steady-state idle connections kept open. Default 5; production is usually 10–50 per worker, calibrated against Postgres's max_connections × number of workers. Over-provisioning kills the database.
• max_overflow — extra connections opened on demand above pool_size. Total cap = pool_size + max_overflow. Set so peak load × workers stays under the database connection cap.
• pool_pre_ping=True — issues a cheap SELECT 1 before handing the connection over. Catches connections silently killed by load balancers (PgBouncer, AWS RDS Proxy, NAT idle timeouts). Without it, the first query of the next request fails with OperationalError: server closed the connection unexpectedly.

engine = create_async_engine(
    DATABASE_URL,
    pool_size=20,
    max_overflow=10,
    pool_pre_ping=True,
    pool_recycle=1800,   # close + reopen every 30 min
)

Behind PgBouncer in transaction mode, set poolclass=NullPool and let PgBouncer pool. Stacking pools is how you exhaust Postgres's max_connections for the whole cluster — every worker holds 30 conns, every PgBouncer client holds another 30, and the database refuses new connections at 8 am Monday.