Vue d'ensemble

SQLAlchemy 2.0, publié le 27 janvier 2023, est une réécriture majeure de l'ORM. Le nouveau style déclaratif avec mapped_column(), le support natif de l'async et une API unifiée modernisent complètement le framework.

Fonctionnalités principales

Nouveau style déclaratif avec mapped_column()

Le nouveau système de mapping utilise Mapped[] et mapped_column() pour définir les colonnes de manière typée. Cela remplace l'ancien Column() et offre une meilleure intégration avec mypy.

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

class Base(DeclarativeBase):
    pass

class Utilisateur(Base):
    __tablename__ = 'utilisateurs'

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

# Le type Python détermine le type SQL et la nullabilité
# Mapped[str] -> NOT NULL, Mapped[str | None] -> NULLABLE

Support natif async

SQLAlchemy 2.0 intègre nativement le support asyncio via create_async_engine et AsyncSession, permettant d'utiliser l'ORM dans des applications asynchrones sans wrapper externe.

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_utilisateurs():
    async with async_session() as session:
        result = await session.execute(
            select(Utilisateur).where(Utilisateur.nom.like('%dupont%'))
        )
        return result.scalars().all()

API select() unifiée

L'ancienne API session.query() est remplacée par le style select() unifié. Toutes les requêtes passent désormais par session.execute(select(...)), offrant une API cohérente entre Core et ORM.

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

with Session(engine) as session:
    # Ancien style (déprécié) : session.query(Utilisateur).filter(...)
    # Nouveau style 2.0 :
    stmt = (
        select(Utilisateur)
        .where(Utilisateur.email.isnot(None))
        .order_by(Utilisateur.nom)
    )
    utilisateurs = session.execute(stmt).scalars().all()

    # Agrégation
    count = session.execute(
        select(func.count()).select_from(Utilisateur)
    ).scalar()

Sources