Using mapped_column() instead of Column() in SQLAlchemy 2.0

Replace id = Column(Integer, primary_key=True) with id: Mapped[int] = mapped_column(primary_key=True) — the Python type and the nullability move into the Mapped[] annotation, and every structural argument stays exactly where it was. This page is the mechanical companion to typed declarative models and Mapped annotations, which covers how the annotation is resolved into a SQL type in the first place.

Quick Answer

Before — legacy declarative:

from sqlalchemy import Column, ForeignKey, Integer, Numeric, String
from sqlalchemy.orm import declarative_base, relationship

Base = declarative_base()


class Order(Base):
    __tablename__ = "orders"

    id = Column(Integer, primary_key=True)
    reference = Column(String(64), nullable=False, unique=True)
    customer_id = Column(Integer, ForeignKey("customers.id"), nullable=False, index=True)
    total = Column(Numeric(12, 2), nullable=False)
    note = Column(String(255), nullable=True)

    customer = relationship("Customer", back_populates="orders")

After — SQLAlchemy 2.0 annotated declarative:

from __future__ import annotations

import decimal

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


class Base(DeclarativeBase):
    type_annotation_map = {str: String(255), decimal.Decimal: Numeric(12, 2)}


class Order(Base):
    __tablename__ = "orders"

    id: Mapped[int] = mapped_column(primary_key=True)
    reference: Mapped[str] = mapped_column(String(64), unique=True)
    customer_id: Mapped[int] = mapped_column(ForeignKey("customers.id"), index=True)
    total: Mapped[decimal.Decimal]
    note: Mapped[str | None]

    customer: Mapped[Customer] = relationship(back_populates="orders")

Both classes produce byte-identical DDL. What changed is that Order.total is now Decimal to a type checker, Order.note is str | None, and nullable= has stopped being a second place where nullability is recorded.

Execution Context & Async Workflow Integration

Nothing about execution changes: mapped_column() produces the same Column object internally, and the statements you build against the converted class are the same statements. What changes is what a type checker knows about the results, which matters most in async code where a mistake surfaces as a runtime exception rather than a wrong value.

# Async — the converted model carries its types through execution
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine

engine = create_async_engine("postgresql+asyncpg://app:secret@db.internal/orders")
Session = async_sessionmaker(engine, expire_on_commit=False)


async def unpaid_total(customer_id: int) -> decimal.Decimal:
    async with Session() as session:
        stmt = select(Order.total).where(Order.customer_id == customer_id)
        result = await session.execute(stmt)
        # scalars() is Sequence[Decimal] to the checker, so this sum is typed
        return sum(result.scalars(), start=decimal.Decimal("0"))

Under the legacy class, Order.total was Any and the start=Decimal("0") argument would have gone unchecked — including the common variant that passes start=0 and produces a TypeError only when the query happens to return rows.

The conversion also interacts with expire_on_commit. A converted model does not change expiry behaviour, but because the attributes are now typed, a None arriving from an expired attribute is a type error your checker can see, where before it was invisible. Async applications generally set expire_on_commit=False on the sessionmaker for the reasons covered in using expire_on_commit=False in FastAPI dependencies.

Resolving Warnings, Errors & Common Mistakes

Advanced Conversion Optimization

Converting by hand is fine for a dozen classes and tedious for two hundred. The mechanical part — moving the type into an annotation and dropping nullable= — is regular enough to script, provided the script refuses to guess.

"""Rewrite legacy Column() declarations to annotated mapped_column().

Deliberately conservative: it converts only the shapes it fully understands and
leaves everything else untouched, with a report of what it skipped. A codemod that
guesses produces a diff nobody can review.
"""
from __future__ import annotations

import re

SIMPLE = re.compile(
    r"^(?P<indent>\s+)(?P<name>\w+)\s*=\s*Column\("
    r"(?P<type>Integer|String\(\d+\)|Numeric\(\d+,\s*\d+\)|Boolean|DateTime)"
    r"(?P<rest>.*)\)\s*$"
)

PY_TYPE = {
    "Integer": "int",
    "Boolean": "bool",
    "DateTime": "datetime.datetime",
}


def convert_line(line: str) -> tuple[str, bool]:
    m = SIMPLE.match(line)
    if not m:
        return line, False
    sql_type, rest = m.group("type"), m.group("rest")
    if sql_type.startswith("String"):
        py = "str"
    elif sql_type.startswith("Numeric"):
        py = "decimal.Decimal"
    else:
        py = PY_TYPE[sql_type]

    nullable = "nullable=True" in rest
    keywords = [
        k.strip() for k in rest.split(",")
        if k.strip() and not k.strip().startswith("nullable=")
    ]
    annotation = f"Mapped[{py} | None]" if nullable else f"Mapped[{py}]"
    call = f"mapped_column({', '.join(keywords)})" if keywords else ""
    suffix = f" = {call}" if call else ""
    return f"{m.group('indent')}{m.group('name')}: {annotation}{suffix}\n", True

Two rules make a codemod like this safe. It must skip anything it does not recognise — a Column with a default= callable, a custom type, an inline ForeignKey with ondelete — and report the skips so they can be done by hand. And its output must be verified by autogenerate rather than by review alone: an empty migration is the only proof that the rewrite preserved the schema. The wider strategy for running this across a large codebase without a long-lived branch is in the legacy 1.4 to 2.0 codemod checklist.

Verifying the Conversion Against the Live Schema

A conversion that changes the schema by accident is worse than no conversion, because the change ships inside a commit whose message says "typing only". The verification is cheap enough to run on every class, and it is the only thing that actually proves the rewrite was faithful.

"""Assert that the models and the live database agree — run this in CI.

Uses Alembic's own comparison machinery rather than reimplementing it, so what the
test compares is exactly what `alembic revision --autogenerate` would compare.
"""
from __future__ import annotations

import asyncio

from alembic.autogenerate import compare_metadata
from alembic.migration import MigrationContext
from sqlalchemy.ext.asyncio import create_async_engine

from myapp.models import Base


async def assert_models_match_database(url: str) -> None:
    engine = create_async_engine(url)
    try:
        async with engine.connect() as connection:
            diff = await connection.run_sync(
                lambda sync_conn: compare_metadata(
                    MigrationContext.configure(sync_conn), Base.metadata
                )
            )
    finally:
        await engine.dispose()

    if diff:
        lines = "\n".join(f"  - {entry}" for entry in diff)
        raise AssertionError(f"models and database disagree:\n{lines}")


if __name__ == "__main__":
    asyncio.run(assert_models_match_database("postgresql+asyncpg://app:secret@localhost/orders"))

Run this against a database migrated to head, in CI, on every pull request. Before a conversion it should already pass; after a conversion it must still pass. When it fails, the diff entries name the table and column, so the cause is usually obvious within a line or two: a column that is nullable in the database and not in the annotation, an unregistered Decimal, or a String whose length in the registry map does not match the column that was actually created years ago.

The last of those is worth expecting. A project that has been running for a while frequently has columns of several different VARCHAR lengths, and a single str: String(255) registry entry will disagree with every column that is not 255. The fix is not to abandon the registry entry but to give the exceptions their own Annotated aliases — str64, str1000 — which documents the exception at the point it applies.

Frequently Asked Questions

Related

• Typed Declarative Models and Mapped Annotations — The parent guide: how annotations resolve to SQL types, and the registry map that makes conversion cheap.
• Typing relationships with Mappedlist... — The relationship half of the conversion, where the annotation picks the collection class.
• Step-by-step guide to SQLAlchemy 2.0 type annotations — The migration-focused walkthrough, including the mypy plugin removal.
• Legacy 1.4 to 2.0 codemod checklist — Running a conversion across a large codebase without a long-lived branch.