Overview

SQLAlchemy 2.0, released on January 27, 2023, is a major ORM rewrite. The new declarative style with mapped_column(), native async support, and a unified API completely modernize the framework.

Main Features

New declarative style with mapped_column()

The new mapping system uses Mapped[] and mapped_column() to define typed columns. This replaces the old Column() and offers better mypy integration.

python 
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from sqlalchemy import String

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = 'users'

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(100))
    email: Mapped[str | None] = mapped_column(String(200))

# The Python type determines the SQL type and nullability
# Mapped[str] -> NOT NULL, Mapped[str | None] -> NULLABLE

Native async support

SQLAlchemy 2.0 natively integrates asyncio support via create_async_engine and AsyncSession, allowing the ORM to be used in asynchronous applications without external wrappers.

python 
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker
from sqlalchemy import select

engine = create_async_engine('sqlite+aiosqlite:///app.db')
async_session = sessionmaker(engine, class_=AsyncSession)

async def get_users():
    async with async_session() as session:
        result = await session.execute(
            select(User).where(User.name.like('%smith%'))
        )
        return result.scalars().all()

Unified select() API

The old session.query() API is replaced by the unified select() style. All queries now go through session.execute(select(...)), providing a consistent API between Core and ORM.

python 
from sqlalchemy import select, func
from sqlalchemy.orm import Session

with Session(engine) as session:
    # Old style (deprecated): session.query(User).filter(...)
    # New 2.0 style:
    stmt = (
        select(User)
        .where(User.email.isnot(None))
        .order_by(User.name)
    )
    users = session.execute(stmt).scalars().all()

    # Aggregation
    count = session.execute(
        select(func.count()).select_from(User)
    ).scalar()

Sources