SQLAlchemy 2.0 & Async Workflows

Modernizing Python Database Interactions

A practical guide to building non-blocking, high-performance database layers with SQLAlchemy’s modern async API.

SQLAlchemy 2.0 & Async Workflows

For years, Python’s asynchronous ecosystem grew rapidly while database ORMs played catch-up. SQLAlchemy 1.4 served as a crucial bridge, but SQLAlchemy 2.0 is the destination. It introduces a unified, forward-compatible API with first-class async support, eliminating the old sync/async divide and giving developers a consistent, type-safe, and performant way to interact with relational databases.

If you’re building modern web services, background workers, or high-concurrency APIs, async SQLAlchemy is no longer optional — it’s the standard. In this article, we’ll walk through setup, real-world workflows, common pitfalls, and when async actually moves the needle.

Why SQLAlchemy 2.0 Changes Everything for Async

Before 2.0, async database access in Python was fragmented. You’d often juggle raw async drivers, write custom session wrappers, or accept partial ORM support. SQLAlchemy 2.0 fixes this by:

• Unifying the API: select(), insert(), update(), and delete() now work identically across sync and async contexts.
• Dropping legacy patterns: session.query() is deprecated in favor of the standalone select() construct.
• Native async drivers: First-class support for asyncpg, aiosqlite, and asyncmy without blocking the event loop.
• Strict typing & modern Python: PEP 484+ type hints, explicit session lifecycle, and predictable coroutine behavior.

The result? A single codebase that can run synchronously during development and asynchronously in production, with minimal friction.

Setting Up Async SQLAlchemy

Async SQLAlchemy requires an async-compatible database driver. For PostgreSQL, that’s asyncpg. For SQLite, aiosqlite.

pip install sqlalchemy asyncpg aiosqlite

Engine & Session Configuration

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

# Async engine (PostgreSQL example)
DATABASE_URL = "postgresql+asyncpg://user:password@localhost:5432/mydb"
engine = create_async_engine(DATABASE_URL, echo=True)

# Async session factory
async_session = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)

Base = DeclarativeBase()

Key notes:

• expire_on_commit=False prevents lazy-loading errors in async contexts.
• Always use async_sessionmaker instead of the sync sessionmaker.
• Connection pooling is handled automatically. Tune pool_size and max_overflow in production.

Writing Async Queries: CRUD in Practice

SQLAlchemy 2.0’s query API is remarkably consistent. Here’s how common operations look in an async workflow:

from sqlalchemy import select, insert, update, delete
from typing import List

# Define a simple model
class User(Base):
    __tablename__ = "users"
    id: int
    name: str
    email: str

# SELECT
async def get_active_users() -> List[User]:
    async with async_session() as session:
        result = await session.scalars(
            select(User).where(User.email != "banned@example.com")
        )
        return result.all()

# INSERT
async def create_user(name: str, email: str) -> User:
    async with async_session() as session:
        new_user = User(name=name, email=email)
        session.add(new_user)
        await session.commit()
        await session.refresh(new_user)
        return new_user

# UPDATE
async def update_user_email(user_id: int, new_email: str) -> bool:
    async with async_session() as session:
        stmt = update(User).where(User.id == user_id).values(email=new_email)
        result = await session.execute(stmt)
        await session.commit()
        return result.rowcount > 0

# DELETE
async def delete_user(user_id: int) -> bool:
    async with async_session() as session:
        stmt = delete(User).where(User.id == user_id)
        result = await session.execute(stmt)
        await session.commit()
        return result.rowcount > 0

Why session.scalars()? It returns ORM instances directly, avoiding the need to unpack Row objects. Use session.execute() when you need raw results, aggregates, or multi-row fetches.

Async Workflows in Production

Real applications don’t just run single queries. They manage transactions, handle errors, and integrate with web frameworks. Here’s how to structure robust async workflows.

Transaction Management

async def transfer_funds(from_id: int, to_id: int, amount: float):
    async with async_session() as session:
        # BEGIN transaction automatically on first operation
        # Use explicit session.begin() if you want finer control
        async with session.begin():
            # Fetch & lock rows (SELECT ... FOR UPDATE)
            from_acc = await session.get(Account, from_id, with_for_update=True)
            to_acc = await session.get(Account, to_id, with_for_update=True)

            if from_acc.balance < amount:
                raise ValueError("Insufficient funds")

            from_acc.balance -= amount
            to_acc.balance += amount

        # Transaction commits on exit of `session.begin()` block
        # Rolls back automatically on exception

FastAPI Integration (Lifespan Pattern)

Modern FastAPI apps should use the lifespan context manager instead of @app.on_event:

from contextlib import asynccontextmanager
from fastapi import FastAPI

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Startup: engine already created globally
    yield
    # Shutdown: dispose engine to close pool connections
    await engine.dispose()

app = FastAPI(lifespan=lifespan)

Dependency injection for sessions:

from fastapi import Depends, HTTPException

async def get_db():
    async with async_session() as session:
        try:
            yield session
        except Exception:
            await session.rollback()
            raise
        finally:
            await session.close()

@app.get("/users")
async def list_users(db: AsyncSession = Depends(get_db)):
    result = await db.scalars(select(User))
    return result.all()

Common Pitfalls & Best Practices

[embed]

Pro Tips:

• Use sqlalchemy.exc.NoResultFound and sqlalchemy.exc.MultipleResultsFound with session.scalar() for single-object lookups.
• Prefer session.get(Model, id) over select().where() for primary key fetches. It’s optimized and returns None if missing.
• Enable echo=False in production. Use structured logging instead.

When Should You Actually Go Async?

Async isn’t a silver bullet. It shines in I/O-bound workloads:

• High-concurrency web APIs (FastAPI, Starlette, Sanic)
• Microservices making parallel external requests + DB calls
• Background task processors with many simultaneous DB operations
• Real-time apps (WebSockets, SSE) requiring persistent DB access

Skip async if:

• Your app is CPU-bound (use multiprocessing or offload to workers)
• You’re running simple CRUD with low concurrency (<100 req/s)
• Your team lacks async debugging experience (it changes error tracing)

Performance gains come from non-blocking event loops, not faster queries. Misconfigured async can actually hurt throughput due to connection pool limits or coroutine overhead. Benchmark your specific workload before migrating.

Conclusion

SQLAlchemy 2.0 brings async database workflows into the modern Python era. The unified API, explicit coroutines, and robust transaction handling make it possible to build high-performance data layers without sacrificing developer experience.

Start by migrating your engine and session factory. Replace session.query() with select(). Add explicit awaits. Integrate with your framework’s async lifecycle. Test under load. The transition is gradual, but the payoff is a codebase that scales cleanly with your traffic.

The era of blocking database calls in async Python is over. SQLAlchemy 2.0 gives you the tools to match it.

For more on how to build SQLAlchemy data layers that hold up in production and Async Workflows.