Implementing SQLAlchemy Async ORM with PostgreSQL: A Practical Guide

Problem and Takeaway

Modern Python services often need non‑blocking database access to keep request latency low. SQLAlchemy 2.0 introduces a fully async ORM that works with the asyncpg driver for PostgreSQL. The key takeaway: create an async engine, use AsyncAttrs on your declarative base, and drive all session work through async with session.begin(). Doing so eliminates the classic “blocking call in event loop” errors and lets you reuse connections from the asyncpg pool.

Prerequisites

• Python 3.11+ (asyncio improvements)
• PostgreSQL 14+ reachable over the network
• Virtual environment with write access to install packages

Install Dependencies

# Run in your project venv
pip install "sqlalchemy[asyncio]" asyncpg

Verify installation:

python -c "import sqlalchemy, asyncpg; print(sqlalchemy.__version__, asyncpg.__version__)"

Create the Async Engine

Use create_async_engine with a PostgreSQL+asyncpg URL. Enable echo=True only for development to see generated SQL.

from sqlalchemy.ext.asyncio import create_async_engine

DATABASE_URL = "postgresql+asyncpg://user:pass@host:5432/dbname"
engine = create_async_engine(DATABASE_URL, echo=False, pool_pre_ping=True)

Windows note: before creating the engine, set the selector event‑loop policy to avoid RuntimeError: Event loop is closed.

import asyncio, sys
if sys.platform.startswith("win"):
    asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())

Define Declarative Models with AsyncAttrs

All ORM classes must inherit from a base that mixes in AsyncAttrs. This enables awaitable attribute access (e.g., await user.orders).

from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from sqlalchemy.ext.asyncio import AsyncAttrs

class Base(AsyncAttrs, DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "users"
    id: Mapped[int] = mapped_column(primary_key=True)
    email: Mapped[str] = mapped_column(unique=True)
    name: Mapped[str]

Create an Async Session Factory

from sqlalchemy.ext.asyncio import async_sessionmaker

async_session = async_sessionmaker(bind=engine, expire_on_commit=False)

expire_on_commit=False keeps instances usable after the transaction ends, which is handy for web handlers that return the object.

Basic CRUD Example

The following self‑contained script demonstrates insert, select, and update using the async session. Run it with python crud_demo.py.

import asyncio
from sqlalchemy import select
from sqlalchemy.exc import IntegrityError

async def main() -> None:
    async with async_session() as session:
        async with session.begin():
            # INSERT
            new_user = User(email="[contact removed]", name="Alice")
            session.add(new_user)
            # Flush to get PK without committing
            await session.flush()
            print(f"Inserted user id={new_user.id}")

        async with session.begin():
            # SELECT
            stmt = select(User).where(User.email == "[contact removed]")
            result = await session.execute(stmt)
            user = result.scalar_one()
            print(f"Queried user: {user.name}")

            # UPDATE
            user.name = "Alice Smith"
            # commit happens on context exit

    # Verify pool reuse
    pool = engine.pool
    print(f"Pool size: {pool.size()}, checked out: {pool.checkedout()}")

if __name__ == "__main__":
    asyncio.run(main())

Common Pitfalls and Diagnostic Decisions

1. Mixing Sync and Async Sessions

Using a regular Session alongside AsyncSession in the same transaction leaks connections because each uses a separate pool. Keep them isolated: either migrate the whole codebase to async or run sync work in a thread pool via asyncio.to_thread.

2. Driver Compatibility

asyncpg requires PostgreSQL 10+ for full feature support. Older servers may reject prepared statements used by the driver. If you see asyncpg.exceptions.InvalidCatalogNameError, verify the server version with SELECT version();.

3. Blocking Calls Inside the Event Loop

Any synchronous I/O (e.g., time.sleep, requests.get) blocks the loop. Replace with async equivalents (await asyncio.sleep, aiohttp). A quick diagnostic: enable engine.echo=True and watch for long pauses between SQL statements.

Verification Steps

1. Round‑trip test – Run the CRUD script above. Assert the printed user id matches the inserted row and the final name is "Alice Smith".
2. SQL logging – Set engine = create_async_engine(..., echo=True). Confirm each logged line starts with "async" (e.g., "async SELECT …") and no "sync" prefixes appear.
3. Pool metrics – After the script finishes, inspect engine.pool.size() and engine.pool.checkedout(). Values should be >0 and checkedout should return to 0, indicating connections are returned to the pool.

Limitations

• Async ORM does not support legacy Query API; you must use select() constructs.
• Some third‑party extensions (e.g., certain Alembic operations) still expect a sync engine. Run migrations with a separate sync engine if needed.
• Connection pool tuning (max_overflow, pool_size) is identical to sync engines but applies to the asyncpg pool; monitor under load.

Rollback / Recovery

If a migration or schema change breaks the async models, you can revert by dropping the affected tables via a sync engine (since DDL is not transactional in PostgreSQL) and re‑applying a known‑good migration. No special async rollback is required because each async with session.begin() block automatically rolls back on exception.

Next Steps

• Integrate the async session into your web framework (FastAPI, Starlette, aiohttp) using a dependency that yields async_session().
• Add retry logic for transient asyncpg.exceptions.ConnectionDoesNotExistError using tenacity or a custom back‑off.
• Benchmark pool settings under realistic concurrency to size pool_size and max_overflow.