Setting Up Async SQLAlchemy

SQLAlchemy 2.0 supports asyncio via create_async_engine (e.g., postgresql+asyncpg://) and async_sessionmaker. Define models with typed Mapped[...] annotations and mapped_column. Use expire_on_commit=False so objects remain usable after commit in async code, configure pool size per worker, and create one session per request via a yield dependency.

from datetime import datetime
from sqlalchemy import ForeignKey, String, func
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship

engine = create_async_engine(settings.database_url, pool_size=10, max_overflow=5, pool_pre_ping=True)
SessionLocal = async_sessionmaker(engine, expire_on_commit=False)


class Base(DeclarativeBase):
    pass


class User(Base):
    __tablename__ = "users"
    id: Mapped[int] = mapped_column(primary_key=True)
    email: Mapped[str] = mapped_column(String(320), unique=True, index=True)
    created_at: Mapped[datetime] = mapped_column(server_default=func.now())
    orders: Mapped[list["Order"]] = relationship(back_populates="user")


class Order(Base):
    __tablename__ = "orders"
    id: Mapped[int] = mapped_column(primary_key=True)
    user_id: Mapped[int] = mapped_column(ForeignKey("users.id", ondelete="CASCADE"), index=True)
    total_cents: Mapped[int]
    user: Mapped[User] = relationship(back_populates="orders")

Common pitfalls

• Sharing one AsyncSession across concurrent tasks — sessions aren't concurrency-safe.
• Default expire_on_commit=True causing lazy reloads (errors in async) after commit.
• Pool size × workers exceeding the database's max connections.

Key points

• create_async_engine + async_sessionmaker
• Typed Mapped models
• One session per request/task

Further reading: SQLAlchemy: asyncio extension