SQLAlchemy est l'un des frameworks de Mapping Objet-Relationnel (ORM) les plus robustes pour Python. Il permet aux développeurs de manipuler des bases de données relationnelles en utilisant des objets et des méthodes Python plutôt que d'écrire manuellement des requêtes SQL complexes. Cette approche facilite la maintenance du code et améliore la portabilité entre différents systèmes de gestion de bases de données (SGBD).

Installation du framework

Pour commencer à utiliser SQLAlchemy, installez le package via l'outil pip :

pip install sqlalchemy

Selon le type de base de données ciblé, l'installation d'un pilote (driver) spécifique est nécessaire :

# Pour PostgreSQL
pip install psycopg2

# Pour MySQL/MariaDB
pip install pymysql

# SQLite est intégré nativement dans la bibliothèque standard de Python.

Initialisation de la connexion

La première étape consiste à configurer l'unité de liaison (Engine) et la fabrique de sessions pour gérer les interactions avec la base de données.

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, declarative_base

# Configuration de l'URL de connexion (exemple avec SQLite)
DATABASE_URL = "sqlite:///ma_boutique.db"

# Initialisation du moteur
db_engine = create_engine(DATABASE_URL, echo=True)

# Création d'une classe de session personnalisée
SessionFactory = sessionmaker(bind=db_engine, autoflush=False, autocommit=False)

# Instance de session active
db_session = SessionFactory()

# Définition de la base pour les modèles
Base = declarative_base()

Modélisation des données et relations

Le système ORM permet de définir les tables sous forme de classes Python. Voici un exemple implémentant une relation de type "Un-à-Plusieurs" et "Plusieurs-à-Plsuieurs".

from sqlalchemy import Column, Integer, String, ForeignKey, Table
from sqlalchemy.orm import relationship

# Table d'association pour la relation Plusieurs-à-Plusieurs
association_articles_tags = Table(
    'association_articles_tags',
    Base.metadata,
    Column('article_id', ForeignKey('articles.id'), primary_key=True),
    Column('tag_id', ForeignKey('tags.id'), primary_key=True)
)

class Utilisateur(Base):
    __tablename__ = 'utilisateurs'
    
    id = Column(Integer, primary_key=True)
    nom_complet = Column(String(100), nullable=False)
    courriel = Column(String(120), unique=True, index=True)
    
    # Relation : Un utilisateur peut avoir plusieurs articles
    articles = relationship("Article", back_populates="auteur")

class Article(Base):
    __tablename__ = 'articles'
    
    id = Column(Integer, primary_key=True)
    titre = Column(String(200), nullable=False)
    contenu = Column(String(2000))
    auteur_id = Column(Integer, ForeignKey('utilisateurs.id'))
    
    auteur = relationship("Utilisateur", back_populates="articles")
    # Relation Plusieurs-à-Plusieurs avec les tags
    tags = relationship("Tag", secondary=association_articles_tags, back_populates="articles")

class Tag(Base):
    __tablename__ = 'tags'
    
    id = Column(Integer, primary_key=True)
    libelle = Column(String(50), unique=True)
    
    articles = relationship("Article", secondary=association_articles_tags, back_populates="tags")

# Génération des tables dans la base de données
Base.metadata.create_all(db_engine)

Opérations de persistance (CRUD)

Insertion de données

# Création d'une nouvelle instance
nouvel_utilisateur = Utilisateur(nom_complet="Jean Dupont", courriel="jean.dupont@email.com")
db_session.add(nouvel_utilisateur)

# Insertion multiple
articles_demo = [
    Article(titre="Introduction à Python", auteur=nouvel_utilisateur),
    Article(titre="Maîtriser les ORM", auteur=nouvel_utilisateur)
]
db_session.add_all(articles_demo)

# Validation des modifications
db_session.commit()

Lecture et filtrage

# Récupération par identifiant unique
user = db_session.query(Utilisateur).get(1)

# Recherche avec filtres complexes
article_python = db_session.query(Article).filter(Article.titre.ilike("%python%")).first()

# Liste de tous les utilisateurs triés par nom
tous_les_users = db_session.query(Utilisateur).order_by(Utilisateur.nom_complet).all()

Mise à jour et suppression

# Modification d'un attribut
article_a_modifier = db_session.query(Article).filter_by(id=1).first()
if article_a_modifier:
    article_a_modifier.titre = "Titre mis à jour"
    db_session.commit()

# Suppression d'un enregistrement
cible_suppression = db_session.query(Utilisateur).filter_by(courriel="jean.dupont@email.com").first()
if cible_suppression:
    db_session.delete(cible_suppression)
    db_session.commit()

Requêtes avancées et agrégations

SQLAlchemy permet de réaliser des jointures et des calculs statistiques via le module func.

from sqlalchemy import func, or_

# Jointure entre Utilisateur et Article
resultats = db_session.query(Utilisateur.nom_complet, Article.titre)\
    .join(Article)\
    .filter(Utilisateur.id == Article.auteur_id)\
    .all()

# Calcul du nombre d'articles par utilisateur
statistiques = db_session.query(
    Utilisateur.nom_complet, 
    func.count(Article.id).label('nb_articles')
).outerjoin(Article).group_by(Utilisateur.nom_complet).all()

# Requête avec condition OR
utilisateurs_filtres = db_session.query(Utilisateur).filter(
    or_(Utilisateur.nom_complet == "Jean Dupont", Utilisateur.courriel.endswith("@pro.com"))
).all()

Gestion sécurisée des sessions

Pour garantir que les ressources sont libérées correctement, il est recommandé d'utiliser un gestionnaire de contexte pour la gestion des sessions.

from contextlib import contextmanager

@contextmanager
def gestion_session():
    session = SessionFactory()
    try:
        yield session
        session.commit()
    except Exception as e:
        session.rollback()
        print(f"Erreur détectée, transaction annulée : {e}")
        raise
    finally:
        session.close()

# Utilisation pratique
with gestion_session() as session:
    nouveau_tag = Tag(libelle="Développement")
    session.add(nouveau_tag)

Principes de conception recommandés

• Injection de dépendances : Gérez vos sessions de manière à ce qu'elles soient facilement injectables dans vos fonctions de service.
• Chargement différé (Lazy Loading) : Soyez vigilant face au problème des requêtes N+1. Utilisez joinedload pour charger les relations nécessaires en une seule requête SQL.
• Séparation des responsabilités : Gardez vos modèles de données distincts de votre logique métier pour une meilleure clarté.
• Migrations : Pour les projets en produtcion, utilisez un outil comme Alembic pour gérer les évolutions de schéma de base de données.