SQLAlchemy 2.0 is much cleaner than 1.x. Mapped types, native async, better cursor semantics. Yet most 2026 codebases use 5% of what’s available. This post is the working set of patterns from real production code.
Modern model definition
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, index=True)
full_name: Mapped[str] = mapped_column(String(120))
is_active: Mapped[bool] = mapped_column(default=True)
created_at: Mapped[datetime] = mapped_column(server_default=func.now())
posts: Mapped[list["Post"]] = relationship(back_populates="author", lazy="selectin")
Mapped[T] types flow through the type checker. Your IDE knows user.email is str.
Sessions and unit of work
Async session pattern:
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
engine = create_async_engine(DSN, pool_size=10, pool_pre_ping=True)
SessionLocal = async_sessionmaker(engine, expire_on_commit=False)
async def get_db():
async with SessionLocal() as session:
yield session
expire_on_commit=False prevents the post-commit re-fetch that’s lethal in async code. See FastAPI + Pydantic v2 + SQLAlchemy 2.0
.
Loading strategies
The classic N+1 trap:
users = await session.scalars(select(User))
for u in users:
print(u.posts) # ⛔ one query per user
Fixes:
selectin: one separate query for posts (preferred, async-friendly):
result = await session.scalars(
select(User).options(selectinload(User.posts))
)
joinedload: a single big JOIN. Good for one-to-one, problematic for one-to-many (cartesian explosion).raiseload: raise instead of fetching. Use in tests to catch accidental laziness.
Set defaults at the relationship level (lazy="selectin" above).
Hybrid properties
Combine Python and SQL into one attribute:
from sqlalchemy.ext.hybrid import hybrid_property
class Order(Base):
subtotal: Mapped[Decimal]
tax_rate: Mapped[Decimal]
@hybrid_property
def total(self) -> Decimal:
return self.subtotal * (Decimal(1) + self.tax_rate)
@total.expression
def total(cls):
return cls.subtotal * (1 + cls.tax_rate)
Now order.total works on instances; select(...).where(Order.total > 100) works on queries. SQL-side and Python-side stay consistent.
Batched inserts
Naive loop:
for row in rows:
session.add(Order(**row))
await session.commit() # 1000 INSERTs
Batched:
await session.execute(
insert(Order),
rows,
)
await session.commit() # 1 INSERT with VALUES
10–100× faster for bulk loads. Use psycopg’s COPY for true bulk (millions of rows).
Upsert (Postgres)
from sqlalchemy.dialects.postgresql import insert as pg_insert
stmt = pg_insert(User).values(...)
stmt = stmt.on_conflict_do_update(
index_elements=[User.email],
set_={"full_name": stmt.excluded.full_name},
)
await session.execute(stmt)
Postgres-flavored upsert. SQLAlchemy 2.0 has clean syntax for it.
Dataclass mixin
from sqlalchemy.orm import MappedAsDataclass
class Base(MappedAsDataclass, DeclarativeBase):
pass
class User(Base):
__tablename__ = "users"
id: Mapped[int] = mapped_column(primary_key=True, init=False)
email: Mapped[str]
full_name: Mapped[str]
u = User(email="[email protected]", full_name="A") # dataclass init
Models become real dataclasses with __init__, __repr__, __eq__.
Raw SQL with parameters
from sqlalchemy import text
result = await session.execute(
text("SELECT id, name FROM users WHERE last_login < :cutoff"),
{"cutoff": datetime.utcnow() - timedelta(days=90)},
)
for row in result:
print(row.id, row.name)
Use named parameters always. Never string-concat.
Migrations with Alembic
alembic init -t async migrations
alembic revision --autogenerate -m "add users table"
alembic upgrade head
Set a naming_convention on Base.metadata so generated names are stable across environments. See FastAPI + Pydantic v2 + SQLAlchemy 2.0
.
Gotchas
1. Mixing sync and async sessions
Don’t share. Async sessions need async engines and async drivers (asyncpg for Postgres).
2. Forgetting to await
session.execute(...) returns a coroutine in async mode. Forgetting await is a silent bug.
3. Long-running sessions
A session that lives for the whole HTTP request is fine. A session that lives across requests holds connections, identity maps, and dirty state. Open per-request via dependency injection.
4. JSON columns without mutable_json_type
Mutating a JSON column in place doesn’t mark the column dirty. Either replace the whole value or use MutableDict / MutableList.
5. Implicit lazy loads in async code
Triggers blocking I/O on attribute access. Use selectinload or raiseload to catch.
Read this next
- FastAPI + Pydantic v2 + SQLAlchemy 2.0 Production Patterns
- PostgreSQL MVCC, Isolation, Locking
- Modern Python Type Hints 2026
- Pydantic v2 Deep Dive
If you want a SQLAlchemy 2.0 + Alembic starter, it’s at rajpoot.dev .
Building something AI-, backend-, or data-heavy and want a second pair of eyes? I do consulting and freelance work — see my projects and ways to reach me at rajpoot.dev .