SQLAlchemy 2.0 marque une rupture majeure avec la version 1.x : l’API Core et ORM ont été refondues pour une meilleure cohérence, les sessions et les transactions sont plus explicites, et le support natif de l’asynchronie via `asyncio` est intégré. En 2026, SQLAlchemy 2.0 est le standard de facto pour l’accès aux bases de données relationnelles en Python, que ce soit pour des scripts simples, des APIs FastAPI/Django ou des pipelines de données. Ce guide pratique couvre les nouveautés de l’API 2.0, les patterns ORM modernes, les migrations avec Alembic et les best practices pour la production.
Nouveautés majeures de SQLAlchemy 2.0
La migration vers SQLAlchemy 2.0 depuis la version 1.4 introduit des changements d’API importants. L’exécution des requêtes se fait maintenant exclusivement via `session.execute()` avec des objets `select()`, `insert()`, `update()` et `delete()` explicites, plutôt que via les méthodes dépréciées `query()`. Cette nouvelle API est plus cohérente entre l’ORM et le Core, et facilite la migration vers des requêtes asynchrones. La méthode `session.query(Model).filter()` est toujours fonctionnelle mais marquée comme dépréciée et sera supprimée dans une version future.
Le typage statique est une amélioration majeure de SQLAlchemy 2.0. Les modèles ORM utilisent maintenant la syntaxe `Mapped[type]` avec les annotations Python de type pour définir les colonnes. Cette approche permet aux outils comme mypy et Pylance de valider statiquement les requêtes et les attributs des modèles, détectant les erreurs de type à l’écriture du code plutôt qu’à l’exécution. `Mapped[Optional[str]]` indique une colonne nullable, `Mapped[str]` une colonne NOT NULL. Cette intégration avec le système de types Python est un changement de paradigme qui améliore significativement la maintenabilité du code.
La gestion des transactions en SQLAlchemy 2.0 est plus explicite et contrôlée. Par défaut, `Session` commence une transaction implicite au premier accès et la conserve jusqu’au `commit()` ou `rollback()`. Le mode « autoflush » envoie automatiquement les changements pendants vers la base de données avant chaque requête, sans les commiter. Pour les cas d’usage avancés, `session.begin()` comme context manager permet un contrôle explicite des transactions, avec rollback automatique en cas d’exception. Cette rigueur dans la gestion des transactions évite les bugs subtils liés aux transactions partielles ou aux commits accidentels.
Définir des modèles ORM modernes avec Mapped
La nouvelle syntaxe de définition des modèles SQLAlchemy 2.0 utilise `DeclarativeBase` et `Mapped`. On crée d’abord une classe de base : `class Base(DeclarativeBase): pass`. Puis chaque modèle hérite de cette base et définit ses colonnes avec la syntaxe `nom_colonne: Mapped[type] = mapped_column(…)`. L’annotation de type dans `Mapped[]` est utilisée à la fois pour le typage statique et pour la définition de la colonne (NULL vs NOT NULL). Cette double utilisation évite la redondance entre la définition de la colonne et son type Python, source d’erreurs dans SQLAlchemy 1.x.
Les relations entre modèles (one-to-many, many-to-many) utilisent également la syntaxe `Mapped`. Une relation one-to-many de User vers Post s’écrit : `posts: Mapped[List[« Post »]] = relationship(back_populates= »author »)`. La notation `List[« Post »]` indique une relation one-to-many, `Optional[« Profile »]` indique une relation one-to-one optionnelle. SQLAlchemy infère automatiquement les paramètres de la relation (foreign key, cascade) depuis ces annotations dans la plupart des cas simples. Pour les cas complexes (relations polymorphiques, self-referential), les paramètres explicites dans `relationship()` restent nécessaires.
Les colonnes avec valeurs par défaut, contraintes et types avancés se définissent dans `mapped_column()`. `mapped_column(default=datetime.utcnow)` pour une valeur par défaut côté Python, `mapped_column(server_default=text(« NOW() »))` pour une valeur par défaut côté base de données. Les index s’ajoutent avec `mapped_column(index=True)` ou des index composites avec `Index(« idx_name », « col1 », « col2 »)` dans la métadonnée de la classe. Les contraintes CHECK, UNIQUE et FOREIGN KEY s’ajoutent dans `__table_args__`. Cette richesse de configuration permet de modéliser fidèlement les contraintes base de données directement dans le code Python.
Requêtes avec la nouvelle API select()
La nouvelle API de requêtage SQLAlchemy 2.0 est centrée sur la fonction `select()`. Une requête simple s’écrit `stmt = select(User).where(User.email == « test@example.com »)` puis `result = session.execute(stmt)` et `users = result.scalars().all()`. La méthode `scalars()` est nécessaire pour extraire les objets ORM des résultats (par opposition aux tuples Row). Pour une requête sur plusieurs colonnes ou plusieurs modèles, on utilise `result.all()` qui retourne des tuples nommés. Cette distinction entre `scalars()` et `all()` est la principale source de confusion pour les développeurs migrant depuis la version 1.x.
Les jointures en SQLAlchemy 2.0 sont construites avec `select().join()`. Une jointure explicite s’écrit `select(Post).join(Post.author)` qui génère un SQL `JOIN` sur la foreign key définie dans la relation. Pour des jointures avec conditions personnalisées, `select(Post).join(User, Post.user_id == User.id)` est l’équivalent. Les jointures gauches utilisent `join(Model, isouter=True)`. Les requêtes avec sous-sélections utilisent `subquery()` et les CTEs récursifs utilisent `cte()`. Ces constructions permettent de générer des SQL arbitrairement complexes tout en restant maintenables grâce à l’abstraction Python.
Le chargement des relations (eager vs lazy loading) est un sujet critique pour les performances des applications SQLAlchemy. Par défaut, les relations sont chargées en « lazy loading » : chaque accès à une relation déclenche une requête SQL supplémentaire, menant au célèbre problème N+1. Pour éviter ce problème, SQLAlchemy 2.0 propose `selectinload()` (charge les relations avec une sous-requête IN), `joinedload()` (charge via JOIN) et `subqueryload()`. On les spécifie dans la requête principale : `select(User).options(selectinload(User.posts))`. Le choix entre ces stratégies dépend du nombre d’objets et de la structure des données.
from sqlalchemy import create_engine, select, String, Integer, ForeignKey
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship, Session
from typing import List, Optional
class Base(DeclarativeBase):
pass
class User(Base):
__tablename__ = "users"
id: Mapped[int] = mapped_column(Integer, primary_key=True)
name: Mapped[str] = mapped_column(String(100), nullable=False)
email: Mapped[str] = mapped_column(String(200), unique=True, nullable=False)
posts: Mapped[List["Post"]] = relationship(back_populates="author", lazy="selectin")
class Post(Base):
__tablename__ = "posts"
id: Mapped[int] = mapped_column(Integer, primary_key=True)
title: Mapped[str] = mapped_column(String(200))
user_id: Mapped[int] = mapped_column(ForeignKey("users.id"))
author: Mapped[Optional["User"]] = relationship(back_populates="posts")
# Exemple requete SQLAlchemy 2.0
engine = create_engine("postgresql+psycopg2://user:pass@localhost/db", echo=True)
with Session(engine) as session:
stmt = select(User).where(User.email.like("%@example.com"))
users = session.execute(stmt).scalars().all()
for user in users:
print(f"{user.name} - {len(user.posts)} posts")SQLAlchemy async : requêtes asynchrones avec asyncio
SQLAlchemy 2.0 intègre nativement le support asyncio via `AsyncSession` et `create_async_engine`. Cette API asynchrone est indispensable pour les applications FastAPI ou Starlette qui reposent sur un event loop asyncio. La configuration utilise un driver asyncio compatible avec la base de données : `asyncpg` pour PostgreSQL, `aiomysql` pour MySQL, `aiosqlite` pour SQLite. Le préfixe `+asyncpg` dans l’URL de connexion active le mode async : `postgresql+asyncpg://user:pass@localhost/mydb`. L’API est quasi-identique à la version synchrone, préfixée par `async`/`await`.
La gestion des sessions asynchrones diffère légèrement de la version synchrone. On utilise `async_sessionmaker` pour créer une factory de sessions, et chaque session est utilisée dans un context manager async : `async with async_session() as session:`. Le `session.execute()` devient `await session.execute()` et le `session.commit()` devient `await session.commit()`. Un piège courant est d’accéder aux relations lazy-loaded depuis du code async : cela lève une `MissingGreenlet` exception. La solution est de toujours utiliser des stratégies d’eager loading (`selectinload`, `joinedload`) dans les requêtes asynchrones.
Pour FastAPI, le pattern recommandé pour les sessions asynchrones utilise une dépendance (`Depends`) qui crée une session par requête HTTP et la ferme automatiquement après. La dépendance utilise `yield` pour permettre le commit en fin de requête et le rollback en cas d’exception. Ce pattern garantit qu’une session n’est jamais partagée entre requêtes (thread-safe) et que les ressources sont toujours libérées. Des bibliothèques comme `fastapi-sqlalchemy` ou `SQLModel` (de Tiangolo, le créateur de FastAPI) encapsulent ce pattern et simplifient l’intégration SQLAlchemy/FastAPI pour les applications CRUD standard.
Migrations avec Alembic
Alembic est l’outil officiel de migrations pour SQLAlchemy. Il génère des scripts Python de migration qui décrivent les changements de schéma de base de données (ajout de colonne, création d’index, modification de contrainte) de manière incrémentielle et versionnée. Chaque migration est un fichier avec deux fonctions : `upgrade()` qui applique le changement et `downgrade()` qui l’annule. Ces fichiers sont committés dans Git avec le code applicatif, garantissant que les changements de schéma sont tracés et reproductibles sur tous les environnements.
Le workflow Alembic standard pour ajouter une fonctionnalité est : modifier les modèles SQLAlchemy dans le code, générer automatiquement une migration avec `alembic revision –autogenerate -m « description »`, vérifier le script généré (Alembic ne détecte pas toujours tous les changements, notamment les renommages), puis appliquer la migration avec `alembic upgrade head`. En CI/CD, on execute `alembic upgrade head` au déploiement pour appliquer les migrations en production. Cette approche permet des déploiements sans intervention manuelle sur la base de données.
Les migrations Alembic peuvent être complexes quand elles impliquent des transformations de données en plus des changements de schéma. Ajouter une colonne NOT NULL à une table existante avec des millions de lignes nécessite une migration en plusieurs étapes : ajouter la colonne nullable, backfiller les valeurs, puis ajouter la contrainte NOT NULL. Faire ces trois opérations en une seule migration peut verrouiller la table pendant des minutes en production. Alembic supporte les migrations « offline » (génération du SQL sans l’exécuter) et les « batches » pour SQLite qui ne supporte pas ALTER TABLE nativement.
Patterns avancés : Repository et Unit of Work
Le pattern Repository encapsule l’accès aux données derrière une interface abstraite, cachant les détails SQLAlchemy au reste de l’application. Au lieu d’appeler directement `session.execute(select(User))` depuis les routes FastAPI, on appelle `user_repo.get_by_email(email)`. Cette abstraction facilite les tests (on peut substituer le repository par un mock en mémoire), améliore la lisibilité des routes (logique métier sans SQL apparent) et centralise les requêtes récurrentes. Le pattern Repository est particulièrement utile dans les architectures hexagonales ou Clean Architecture où le domaine métier doit être indépendant de la couche de persistance.
Le pattern Unit of Work fonctionne de pair avec le Repository pour coordonner les transactions qui impliquent plusieurs aggregats. Dans SQLAlchemy, la Session est déjà une implémentation du Unit of Work : elle maintient un registre des objets chargés et trackés (identity map), détecte automatiquement les changements (dirty tracking) et les synchronise avec la base de données au flush. Expliciter ce pattern dans le code en créant une classe `UnitOfWork` qui encapsule la session et expose des repositories est une pratique répandue dans les applications complexes pour rendre les intentions transactionnelles claires dans le code.
Les événements SQLAlchemy permettent de brancher des hooks sur le cycle de vie des objets ORM. `@event.listens_for(User, « before_insert »)` déclenche un callback avant chaque insertion d’un User. Ces hooks sont utiles pour des besoins transversaux : horodatage automatique (`created_at`, `updated_at`), audit logs, validation des données avant persistence, ou synchronisation avec des caches externes. En SQLAlchemy 2.0, les mixins avec des colonnes d’audit (`TimestampMixin` avec `created_at` et `updated_at`) utilisant `mapped_column(default=func.now(), onupdate=func.now())` sont la façon idiomatique de gérer ces métadonnées communes à de nombreux modèles.
Performance et optimisation des requêtes
L’analyse des requêtes SQL générées par SQLAlchemy est indispensable pour identifier les problèmes de performance. En mode debug, SQLAlchemy peut logger toutes les requêtes SQL avec `echo=True` dans `create_engine()`. En production, des outils comme `py-spy` ou `sqlalchemy-query-counter` permettent de détecter les requêtes N+1 et les requêtes trop lentes. La méthode `explain()` ou `EXPLAIN ANALYZE` au niveau SQL révèle si les index sont utilisés correctement. Une requête qui fait un scan séquentiel sur une grande table signale souvent un index manquant ou une requête mal formulée.
Le connection pooling est automatiquement géré par SQLAlchemy via `QueuePool`. En production, configurer correctement la taille du pool est crucial : `pool_size=5, max_overflow=10` signifie 5 connexions permanentes et 10 supplémentaires en cas de pic. `pool_timeout=30` évite les blocages indéfinis. `pool_pre_ping=True` vérifie que les connexions du pool sont toujours valides avant utilisation, évitant les erreurs « Connection closed » après une coupure réseau ou un redémarrage de la base de données. Ces paramètres doivent être ajustés selon la charge de l’application et le nombre maximum de connexions autorisées par le serveur de base de données.
Les requêtes en bulk (insertions et mises à jour massives) doivent éviter les ORM objects et utiliser l’API Core pour les performances. `session.execute(insert(User), [{« name »: « Alice »}, {« name »: « Bob »}])` génère un seul INSERT avec de multiples valeurs, bien plus efficace que de créer et d’ajouter des objets User un par un. Pour les updates en masse, `session.execute(update(User).where(User.active == False).values(deleted_at=func.now()))` génère un UPDATE SQL direct sans charger les objets en mémoire. Ces optimisations peuvent réduire le temps d’exécution d’opérations en masse de plusieurs ordres de grandeur par rapport aux patterns ORM naïfs.
Tests avec SQLAlchemy et bases de données
Tester des applications SQLAlchemy peut se faire avec une base de données de test réelle (recommandé pour la fiabilité) ou avec SQLite en mémoire (rapide mais avec des différences comportementales). L’approche recommandée en 2026 est d’utiliser PostgreSQL dans Docker pour les tests d’intégration, avec une base de données créée et détruite pour chaque suite de tests. Des frameworks comme pytest avec pytest-asyncio pour les tests async et des fixtures qui initialisent le schéma avec `Base.metadata.create_all(engine)` au setup rendent cette approche pratique.
Le pattern de transaction rollback pour les tests isolés évite de repartir d’une base vide pour chaque test tout en garantissant l’isolation. On démarre une transaction au début de chaque test, on exécute le test dans cette transaction, puis on rollback à la fin plutôt que de commiter. La base reste dans son état initial pour le test suivant. Ce pattern est implémenté avec des savepoints SQLAlchemy pour les tests qui appellent du code applicatif qui lui-même commit des transactions, ce qui nécessite une configuration de la session de test pour utiliser les savepoints plutôt que de vraies transactions.
Les factories d’objets de test avec une bibliothèque comme `factory_boy` simplifient la création de données de test réalistes pour les tests SQLAlchemy. On définit une factory pour chaque modèle avec des valeurs par défaut sensées et des générateurs aléatoires pour les champs uniques (email, username). Dans les tests, on crée des objets avec `UserFactory.create(name= »Alice »)` qui les insère automatiquement en base. Les factories peuvent gérer les dépendances entre objets (créer automatiquement un User quand on crée un Post qui lui est associé), réduisant significativement le boilerplate de setup des tests.
Commentaires (0)
Laisser un commentaire
Les commentaires sont modérés. Questions WordPress, cybersécurité ou dev web bienvenues.