🐍 SQLAlchemy – ORM Python & Architecture de données

Les grandes couches de SQLAlchemy

Cycle de vie simplifié d’une requête ORM

                    [Application Web]   (FastAPI, Flask...)
                    │
                    ▼
                    Création / injection d'une Session (scopée à la requête HTTP)
                    │
                    ▼
                    ORM :
                    - construction d'un select() / query
                    - résolution des relationships / lazy loading
                    │
                    ▼
                    Core :
                    - génération du SQL adapté au dialecte
                    - envoi au Engine
                    │
                    ▼
                    Engine / Driver :
                    - obtention d'une connexion depuis le pool
                    - exécution réelle du SQL
                    │
                    ▼
                    Résultats :
                    - mapping lignes ➜ objets Python
                    - Session suit les changements (dirty, new, deleted)
                    │
                    ▼
                    Fin de requête :
                    - commit() ou rollback()
                    - fermeture / retour de la connexion au pool
                

Construire une URL de base de données

Créer un Engine & comprendre le pooling

                    from sqlalchemy import create_engine

                    DATABASE_URL = "postgresql+psycopg://app:secret@localhost:5432/appdb"

                    engine = create_engine(
                    DATABASE_URL,
                    echo=True,          # log SQL dans la console
                    pool_size=10,       # connexions simultanées
                    max_overflow=20,    # connexions additionnelles
                    pool_pre_ping=True  # vérifie que la connexion est vivante
                    )
                

Bonnes pratiques :

• Un engine global par process (ou par BDD), partagé dans tout le projet.
• Garder l’URL dans les variables d’environnement.
• Activer echo=True uniquement en dev (bruyant).

Types & contraintes usuels

MetaData & création des tables

                    from sqlalchemy import create_engine

                    engine = create_engine(DATABASE_URL)

                    # crée toutes les tables connues par Base.metadata
                    Base.metadata.create_all(engine)

                    # pour drop (danger !)
                    # Base.metadata.drop_all(engine)
                

En pratique, on utilise plutôt Alembic pour gérer l’évolution du schéma au fil du temps, mais create_all() est pratique pour des POC ou des tests.

Unit of Work : suivre les changements

                    with SessionLocal() as session:
                    user = User(email="john@example.com")
                    session.add(user)      # new
                    user.is_active = False # dirty
                    session.commit()       # flush + commit transaction
                

La Session garde en mémoire :

• new : objets nouvellement ajoutés.
• dirty : objets modifiés.
• deleted : objets marqués pour suppression.

Pattern “Session par requête HTTP”

                    # Exemple FastAPI
                    from fastapi import Depends, FastAPI
                    from sqlalchemy.orm import Session

                    app = FastAPI()

                    def get_db() -> Session:
                    db = SessionLocal()
                    try:
                    yield db
                    db.commit()
                    except:
                    db.rollback()
                    raise
                    finally:
                    db.close()

                    @app.get("/users")
                    def list_users(db: Session = Depends(get_db)):
                    return db.query(User).all()
                

On lie le cycle de vie de la Session au cycle de vie HTTP : ouverture en début de requête, commit/rollback en fin, puis close.

Many-to-Many via table d’association

                    from sqlalchemy import Table, Column

                    article_tag = Table(
                    "article_tag",
                    Base.metadata,
                    Column("article_id", ForeignKey("articles.id"), primary_key=True),
                    Column("tag_id", ForeignKey("tags.id"), primary_key=True),
                    )

                    class Article(Base):
                    __tablename__ = "articles"
                    id: Mapped[int] = mapped_column(primary_key=True)
                    title: Mapped[str]

                    tags: Mapped[list["Tag"]] = relationship(
                    secondary=article_tag, back_populates="articles"
                    )

                    class Tag(Base):
                    __tablename__ = "tags"
                    id: Mapped[int] = mapped_column(primary_key=True)
                    name: Mapped[str]

                    articles: Mapped[list["Article"]] = relationship(
                    secondary=article_tag, back_populates="tags"
                    )
                

Gestion du chargement : lazy vs eager

• lazy="select" (par défaut) : une requête par collection (risque N+1).
• selectinload : charge toutes les relations en un second SELECT IN.
• joinedload : fait une jointure SQL (JOIN).

                    from sqlalchemy.orm import selectinload, joinedload
                    from sqlalchemy import select

                    stmt = (
                    select(User)
                    .options(selectinload(User.posts))
                    .order_by(User.id)
                    )
                    users = session.scalars(stmt).all()
                

Pattern “app context” (Flask)

                    from flask import Flask, g
                    from sqlalchemy.orm import Session

                    app = Flask(__name__)

                    def get_db() -> Session:
                    if "db" not in g:
                    g.db = SessionLocal()
                    return g.db

                    @app.teardown_appcontext
                    def teardown_db(exc):
                    db = g.pop("db", None)
                    if db is not None:
                    if exc is None:
                    db.commit()
                    else:
                    db.rollback()
                    db.close()

                    @app.route("/users")
                    def users_view():
                    db = get_db()
                    users = db.query(User).all()
                    return {"users": [u.email for u in users]}
                

On stocke la Session dans g (contexte request), puis on commit/rollback dans un teardown_appcontext.

Utiliser SQLAlchemy dans Django & micro-services

                            # ex: service réutilisable
                            from myproject.sa import SessionLocal

                            def count_orders_by_status():
                            with SessionLocal() as session:
                            return (
                            session.query(Order.status, func.count(Order.id))
                            .group_by(Order.status)
                            .all()
                            )
                        

Index et optimisation des requêtes

                            from sqlalchemy import Index

                            class User(Base):
                            __tablename__ = "users"
                            id: Mapped[int] = mapped_column(primary_key=True)
                            email: Mapped[str] = mapped_column(String(255), unique=True, index=True)
                            is_active: Mapped[bool] = mapped_column(index=True)

                            __table_args__ = (
                            Index("idx_user_active_email", "is_active", "email"),
                            )
                        

Bulk operations & tuning avancé

                    # Bulk insert (Core)
                    from sqlalchemy import insert

                    rows = [
                    {"email": f"user{i}@example.com"}
                    for i in range(10_000)
                    ]

                    with engine.begin() as conn:
                    conn.execute(insert(User), rows)

                    # ORM : batch + flush par paquets
                    batch_size = 1000
                    with SessionLocal() as session:
                    for i in range(0, len(rows), batch_size):
                    for row in rows[i:i+batch_size]:
                    session.add(User(**row))
                    session.flush()   # envoie le batch
                    session.commit()
                

• Utiliser Core pour les très gros imports (plus proche du driver).
• Toujours regrouper les écritures dans une transaction unique.
• Éviter de charger en mémoire des millions de lignes : streamer les résultats.

Filtres, agrégats, pagination

                    from sqlalchemy import func

                    # Nombre d'utilisateurs actifs
                    count_stmt = select(func.count(User.id)).where(User.is_active == True)
                    active_count = session.scalar(count_stmt)

                    # Pagination simple
                    page = 1
                    page_size = 20

                    stmt = (
                    select(Post)
                    .order_by(Post.created_at.desc())
                    .offset((page-1)*page_size)
                    .limit(page_size)
                    )
                    posts = session.scalars(stmt).all()
                

Utiliser Core directement (sans ORM)

                    from sqlalchemy import Table, Column, Integer, String, MetaData, select

                    metadata = MetaData()

                    user_table = Table(
                    "users",
                    metadata,
                    Column("id", Integer, primary_key=True),
                    Column("email", String(255)),
                    )

                    with engine.connect() as conn:
                    result = conn.execute(select(user_table).where(user_table.c.id == 1))
                    row = result.one()
                

Créer & appliquer des révisions

                    # Génération automatique en comparant Base.metadata et la BDD
                    alembic revision --autogenerate -m "create user & post tables"

                    # Appliquer les migrations
                    alembic upgrade head

                    # Revenir en arrière
                    alembic downgrade -1
                

Workflow typique équipe

1. Modifier les modèles (classes Python).
2. Lancer alembic revision --autogenerate et relire le SQL généré.
3. Commit du code + migration dans Git.
4. En recette / prod : alembic upgrade head dans le pipeline CI/CD.