SQLAlchemy : ORM Python pour les applications data-driven

SQLAlchemy est l’ORM de référence Python. Il offre deux niveaux d’abstraction — le Core (SQL expression language) et l’ORM — et supporte pleinement l’async depuis la version 1.4.

Modèles déclaratifs

from datetime import datetime
from sqlalchemy import String, Integer, ForeignKey, DateTime, Enum
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship
import enum

class Base(DeclarativeBase):
    pass

class OrderStatus(enum.Enum):
    PENDING = "pending"
    CONFIRMED = "confirmed"
    SHIPPED = "shipped"
    DELIVERED = "delivered"

class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(primary_key=True)
    email: Mapped[str] = mapped_column(String(255), unique=True, nullable=False)
    name: Mapped[str] = mapped_column(String(100))
    created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow)

    # Relation one-to-many
    orders: Mapped[list["Order"]] = relationship(back_populates="user",
                                                  cascade="all, delete-orphan")

class Order(Base):
    __tablename__ = "orders"

    id: Mapped[int] = mapped_column(primary_key=True)
    user_id: Mapped[int] = mapped_column(ForeignKey("users.id"), nullable=False)
    total: Mapped[float] = mapped_column(nullable=False)
    status: Mapped[OrderStatus] = mapped_column(Enum(OrderStatus),
                                                  default=OrderStatus.PENDING)
    created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow)

    user: Mapped["User"] = relationship(back_populates="orders")
    items: Mapped[list["OrderItem"]] = relationship(back_populates="order")

Sessions et requêtes (version async)

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker
from sqlalchemy import select, func

engine = create_async_engine(
    "postgresql+asyncpg://user:password@localhost/mydb",
    echo=False,
    pool_size=10,
    max_overflow=20
)
async_session = async_sessionmaker(engine, expire_on_commit=False)

async def get_user_with_orders(user_id: int) -> User | None:
    async with async_session() as session:
        stmt = (
            select(User)
            .where(User.id == user_id)
            .options(selectinload(User.orders))   # eager loading évite le N+1
        )
        result = await session.execute(stmt)
        return result.scalar_one_or_none()

async def get_top_customers(limit: int = 10) -> list[tuple]:
    async with async_session() as session:
        stmt = (
            select(User.name, func.sum(Order.total).label("total_spent"))
            .join(Order)
            .where(Order.status == OrderStatus.DELIVERED)
            .group_by(User.id, User.name)
            .order_by(func.sum(Order.total).desc())
            .limit(limit)
        )
        result = await session.execute(stmt)
        return result.all()

Transactions et gestion des erreurs

async def create_order(user_id: int, items: list[dict]) -> Order:
    async with async_session() as session:
        async with session.begin():    # transaction automatique
            try:
                user = await session.get(User, user_id)
                if not user:
                    raise ValueError(f"User {user_id} not found")

                order = Order(
                    user_id=user_id,
                    total=sum(i['price'] * i['quantity'] for i in items),
                    status=OrderStatus.PENDING
                )
                session.add(order)
                await session.flush()   # obtenir l'ID sans commit

                for item_data in items:
                    item = OrderItem(order_id=order.id, **item_data)
                    session.add(item)

                # session.begin() commit automatiquement à la sortie du with
                return order

            except Exception:
                # session.begin() rollback automatiquement en cas d'exception
                raise

Migrations avec Alembic

# Initialiser Alembic
alembic init alembic

# Générer une migration depuis les changements de modèles
alembic revision --autogenerate -m "add orders table"

# Appliquer les migrations
alembic upgrade head

# Rollback d'une migration
alembic downgrade -1

# alembic/env.py — connecter à vos modèles
from myapp.models import Base
target_metadata = Base.metadata

Notre formation Python couvre SQLAlchemy et la gestion des bases de données avec des projets concrets.