SQLAlchemy est une bibliothèque ORM (Object-Relational Mapper) extrêmement robuste et flexible pour Python, conçue pour opérer avec les bases de données d'une manière orientée objet. Elle englobe des fonctionnalités avancées telles que la gestion des pools de connexions, le support des transactions et la capacité à exécuter des requêtes complexes, améliorant considérablement la manipulation des données depuis le code Python. L'adoption d'un ORM plutôt que l'accès direct via SQL brut est une pratique courante dans le développement moderne, offrant une abstraction plus élevée et une meilleure maintenabilité.

1. Initialisation du Moteur de Base de Données

Toute application utilisant SQLAlchemy démarre avec la création d'un objet Engine. Cet objet sert de point central pour toutes les interactions avec une base de données spécifique, fournissant une fabrique de connexions via un pool de connexions intégré. Il est généralement instancié une seule fois par serveur de base de données et utilisé globalement dans l'application.

from sqlalchemy import create_engine, text, Column, Integer, String, TIMESTAMP, ForeignKey, Index, UniqueConstraint, func
from sqlalchemy.orm import declarative_base, relationship, sessionmaker, scoped_session
from urllib.parse import quote_plus
import contextlib

# Paramètres de connexion hypothétiques
NOM_UTILISATEUR = "utilisateur_db"
MOT_DE_PASSE = "motdepasse_securise@123" # Exemple avec caractère spécial
HOTE_DB = "localhost"
PORT_DB = 3306
NOM_BASE_DE_DONNEES = "application_db"

# Construction de la chaîne de connexion
# Pour les caractères spéciaux dans le mot de passe, utilisez quote_plus
chaine_connexion_base = f"mysql+pymysql://{NOM_UTILISATEUR}:{MOT_DE_PASSE}@{HOTE_DB}:{PORT_DB}/{NOM_BASE_DE_DONNEES}?charset=utf8"
chaine_connexion_encodee = f"mysql+pymysql://{NOM_UTILISATEUR}:{quote_plus(MOT_DE_PASSE)}@{HOTE_DB}:{PORT_DB}/{NOM_BASE_DE_DONNEES}?charset=utf8"

# Création du moteur
# pool_size : Nombre de connexions maintenues dans le pool.
# pool_recycle : Durée en secondes après laquelle une connexion est recyclée.
# echo=True affichera toutes les requêtes SQL exécutées dans la console.
moteur_db_app = create_engine(chaine_connexion_encodee, pool_size=15, pool_recycle=3600, echo=False)
print("Moteur de base de données initialisé.")

L'objet Engine est chargé de manière "paresseuse" (lazy-loaded). Cela signifie qu'aucune connexion réelle à la base de données n'est établie au moment de l'appel à create_engine(). Les connexions sont demandées et établies uniquement lorsqu'une opération sur la base de données est initiée, par exemple via Engine.connect() ou Engine.execute().

2. Définitoin des Modèles ORM

L'essence de l'ORM réside dans la représentation des entités de la base de données (tables, colonnes) comme des objets Python. SQLAlchemy offre plusieurs approches pour déclarer des modèles de données, notamment MetaData(), registry() et declarative_base(). Nous nous concentrerons sur declarative_base(), une méthode très pratique et largement adoptée.

Définition Simple d'un Modèle

Voici comment définir un modèle simple pour une table d'articles, incluant une clé primaire et des champs de texte.

BaseDeclerative = declarative_base()

class Article(BaseDeclerative):
    __tablename__ = 'articles' # Nom de la table dans la base de données
    identifiant_article = Column(Integer, primary_key=True, autoincrement=True, comment='Clé primaire, identifiant unique de l\'article')
    titre_article = Column(String(120), nullable=False, comment='Titre de l\'article, ne peut être nul')
    contenu_principal = Column(String(16000), comment='Contenu détaillé de l\'article')
    date_creation = Column(TIMESTAMP, server_default=text('CURRENT_TIMESTAMP'), comment='Date et heure de création de l\'article')

    # Relation avec les propriétés de l'article (sera définie plus bas)
    details_article = relationship('ProprietesArticle', backref='article_associe', uselist=False, lazy='select')

    # Déclaration d'un index sur la date de création
    __table_args__ = (
        Index('idx_date_creation_desc', date_creation.desc()),
    )

class ProprietesArticle(BaseDeclerative):
    __tablename__ = 'proprietes_articles'
    identifiant_prop = Column(Integer, primary_key=True, autoincrement=True, comment='Clé primaire des propriétés')
    fk_article_id = Column(Integer, ForeignKey('articles.identifiant_article'), nullable=False, comment='Clé étrangère vers l\'article parent')
    votes_total = Column(Integer, default=5, server_default='5', comment='Nombre de votes pour l\'article')
    vues_total = Column(Integer, default=100, server_default='100', comment='Nombre de vues de l\'article')
    derniere_maj = Column(TIMESTAMP, server_default=text('CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP'), comment='Dernière mise à jour de ces propriétés')

    # Contrainte d'unicité sur la clé étrangère pour garantir une seule entrée par article
    __table_args__ = (
        UniqueConstraint('fk_article_id', name='uc_fk_article_prop'),
    )

# Création des tables dans la base de données
BaseDeclerative.metadata.create_all(moteur_db_app)
print("Modèles de tables créés dans la base de données.")

Attributs des Colonnes

Les objets Column acceptent divers paramètres pour définir les caractéristiques des champs de la base de données :

• nullable=False : rend le champ non nul.
• comment='...' : ajoute un commentaire pour la colonne.
• default=... : définit une valeur par défaut côté Python pour les nouvelles instances de modèle.
• server_default=... : définit une valeur par défaut côté base de données, souvent utilisée avec text() ou func() pour des valeurs dynamiques (comme CURRENT_TIMESTAMP).

La distinction clé entre default et server_default est que le premier est appliqué par SQLAlchemy avant l'envoi à la base de données, tandis que le second est géré directement par le SGBD.

Pour les valeurs par défaut de type horodatage, on utilise souvent text('CURRENT_TIMESTAMP') ou func.now(). Pour MySQL, la clause ON UPDATE CURRENT_TIMESTAMP doit être spécifiée directement dans server_default via text(), car server_onupdate n'est pas supporté pour cet usage.

Relations, Contraintes et Index

La définition des relations entre les tables, des contraintes d'intégrité et des index est cruciale pour la performance et la cohérence des données.

• ForeignKey('nom_table.nom_colonne') : Établit une clé étrangère.
• relationship('NomDuModeleCible') : Définit la relation entre les modèles ORM.
• Index('nom_index', colonne.desc()) : Crée un index sur une colonne ou plusieurs, potentiellement avec un ordre spécifique.
• UniqueConstraint('colonne', name='nom_contrainte') : Ajoute une contrainte d'unicité.

Il est important de noter que si SQLAlchemy permet de définir la structure de la base de données via les modèles, il n'est généralement pas recommandé de l'utiliser pour gérer les modifications de schéma (migrations) en production. Des outils dédiés comme Alembic sont préférables pour cette tâche. L'ORM est avant tout un outil de gestion des données, pas de la structure.

3. Gestion des Sessions

La Session est un concept fondamental dans SQLAlchemy, agissant comme l'interface principale pour interagir avec les objets de modèle via le moteur de base de données.

Création d'une Session

La méthode recommandée pour créer des sessions est d'utiliser sessionmaker, qui configure une fabrique de sessions liée à un Engine spécifique.

FabriqueSession = sessionmaker(bind=moteur_db_app)

session_directe_1 = FabriqueSession()
session_directe_2 = FabriqueSession()
print(f"\nSessions créées directement : session_directe_1 est session_directe_2 -> {session_directe_1 is session_directe_2}") # Devrait être False

Comme le montre l'exemple, chaque appel à FabriqueSession() retourne une nouvelle instance de session. Dans un environnement multithread ou Web, cela peut mener à des problèmes de cohérence et de gestion des ressources.

Sessions Scoped (Scoped Sessions)

Pour les applications qui nécessitent une session unique par contexte (par exemple, par thread ou par requête Web), scoped_session est la solution recommandée. Elle fournit une session "globale" qui est en fait stockée dans un contexte spécifique (généralement le thread actuel), garantissant une session unique par contexte.

SessionEnPortee = scoped_session(FabriqueSession)

session_scope_1 = SessionEnPortee()
session_scope_2 = SessionEnPortee()
print(f"Sessions en portée (scoped) : session_scope_1 est session_scope_2 -> {session_scope_1 is session_scope_2}") # Devrait être True

# Il est impératif de retirer la session à la fin de son cycle de vie (ex: fin de requête web)
SessionEnPortee.remove()
print("Session en portée retirée.")

Gestion des Sessions dans les Applications Web

L'utilisation de scoped_session dans une application Web exige deux points importants :

1. L'enregistrement de l'objet scoped_session au démarrage de l'application, le rendant accessible à toutes les parties de l'application.
2. L'appel à scoped_session.remove() à la fin de chaque requête Web. Cela garantit que la session est correctement fermée et que les ressources sont libérées. Cette opération est souvent intégrée aux événements de fin de requête du framework Web utilisé (par exemple, Flask, Django).

Pour simplifier, de nombreux frameworks Web offrent des extensions qui intègrent SQLAlchemy et gèrent automatiquement le cycle de vie des sessions (ex: Flask-SQLAlchemy).

4. Manipulation des Données (CRUD)

SQLAlchemy offre des méthodes intuitives pour insérer, modifier et supprimer des données, tant pour des opérations unitaires que pour des traitements de masse.

Insertion de Données

Pour insérer un nouvel enregistrement, créez une instance du modèle, ajoutez-la à la session, puis appelez commit(). Les relations sont gérées de manière transparente.

session_actuelle = SessionEnPortee() # Obtenir une session depuis la portée
try:
    nouvel_article = Article(
        titre_article="Mon Premier Article ORM",
        contenu_principal="Ceci est le contenu détaillé de mon tout premier article via ORM.",
        details_article=ProprietesArticle(votes_total=15, vues_total=250)
    )
    session_actuelle.add(nouvel_article)
    session_actuelle.commit()
    print(f"\nArticle inséré avec succès. ID: {nouvel_article.identifiant_article}, Titre: {nouvel_article.titre_article}")

    # Exemple de flush vs commit
    article_en_attente = Article(titre_article="Article en attente de commit", contenu_principal="Contenu qui sera flushed mais pas commité.")
    session_actuelle.add(article_en_attente)
    session_actuelle.flush() # L'ID sera généré, mais pas de commit réel à la DB
    print(f"Article en attente flushed. ID généré: {article_en_attente.identifiant_article}")
    session_actuelle.rollback() # Annule l'opération après le flush
    print(f"Article en attente annulé. Il n'est pas dans la base de données.")

except Exception as e:
    session_actuelle.rollback()
    print(f"Erreur lors de l'insertion : {e}")
finally:
    SessionEnPortee.remove()

session.add() place l'objet dans la session en attente. session.flush() envoie les modifications en attente à la base de données mais ne les rend pas permanentes (la transaction reste ouverte). session.commit() rend les modifications permanentes et clôture la transaction. Un commit() exécute implicitement un flush() au préalable.

Modification et Suppression de Données

La modification et la suppression suivent un modèle similaire : récupérer l'objet ou les objets, puis appliquer l'opération.

session_actuelle = SessionEnPortee()
try:
    # Insérer un article pour les exemples de modification/suppression
    article_test_crud = Article(titre_article="Article pour CRUD", contenu_principal="Contenu pour les tests CRUD.")
    session_actuelle.add(article_test_crud)
    session_actuelle.commit()
    id_article_crud = article_test_crud.identifiant_article

    # Modifier un article
    # synchronize_session="fetch" assure que les objets affectés sont rechargés depuis la DB
    session_actuelle.query(Article).filter(Article.identifiant_article == id_article_crud).update(
        {"titre_article": "Article CRUD Modifié", "contenu_principal": "Nouveau contenu mis à jour."},
        synchronize_session="fetch"
    )
    session_actuelle.commit()
    print(f"Article {id_article_crud} modifié avec succès.")

    # Supprimer un article
    session_actuelle.query(Article).filter(Article.identifiant_article == id_article_crud).delete(
        synchronize_session="fetch"
    )
    session_actuelle.commit()
    print(f"Article {id_article_crud} supprimé avec succès.")

except Exception as e:
    session_actuelle.rollback()
    print(f"Erreur lors des opérations CRUD : {e}")
finally:
    SessionEnPortee.remove()

Le paramètre synchronize_session="fetch" est important pour s'assurer que les objets dans la session sont synchronisés avec l'état réel de la base de données après une mise à jour ou une suppression, évitant ainsi des incohérences si des validations sont nécessaires côté base de données.

Insertion de Données en Masse

Pour les volumes importants de données, SQLAlchemy propose des méthodes optimisées pour l'insertion en masse, évitant la surcharge de l'ORM.

session_actuelle = SessionEnPortee()
try:
    # Liste d'objets pour insertion en masse
    liste_objets_articles = [
        Article(titre_article=f'Article de masse {i}', contenu_principal=f'Contenu généré pour le lot {i}')
        for i in range(1, 101) # 100 articles
    ]

    # Liste de dictionnaires pour insertion en masse (plus rapide)
    liste_dictionnaires_articles = [
        {"titre_article": f"Article par dict {i}", "contenu_principal": f"Contenu du dictionnaire {i}"}
        for i in range(101, 201) # 100 articles
    ]

    # Méthode 1: add_all (prend des objets, supporte les relations, mais plus lente)
    # session_actuelle.add_all(liste_objets_articles)
    # session_actuelle.commit()
    # print(f"{len(liste_objets_articles)} articles insérés via add_all.")

    # Méthode 2: bulk_save_objects (prend des objets, ne supporte pas les relations, plus rapide)
    # Ne pas inclure de relations dans les objets pour cette méthode
    # session_actuelle.bulk_save_objects(liste_objets_articles)
    # session_actuelle.commit()
    # print(f"{len(liste_objets_articles)} articles insérés via bulk_save_objects.")

    # Méthode 3: bulk_insert_mappings (prend des dictionnaires, plus rapide)
    session_actuelle.bulk_insert_mappings(Article, liste_dictionnaires_articles)
    session_actuelle.commit()
    print(f"{len(liste_dictionnaires_articles)} articles insérés via bulk_insert_mappings.")

    # Méthode 4: session.execute(insert(), [dictionnaires]) (la plus rapide)
    liste_dictionnaires_execute = [
        {"titre_article": f"Article par execute {i}", "contenu_principal": f"Contenu exécuté {i}"}
        for i in range(201, 301) # 100 articles
    ]
    session_actuelle.execute(Article.__table__.insert(), liste_dictionnaires_execute)
    session_actuelle.commit()
    print(f"{len(liste_dictionnaires_execute)} articles insérés via session.execute(insert()).")

except Exception as e:
    session_actuelle.rollback()
    print(f"Erreur lors de l'insertion en masse : {e}")
finally:
    SessionEnPortee.remove()

En général, session.execute(Article.__table__.insert(), ...) et bulk_insert_mappings sont les plus performantes pour les insertions en masse, mais elles ne gèrent pas les relations ORM. add_all et bulk_save_objects supportent les objets ORM mais sont plus lentes. Le choix dépend du scénario d'utilisation (avec ou sans relations, performance critique).

5. Requêtes de Données

SQLAlchemy offre un éventail complet de méthodes pour interroger la base de données, allant de la récupération de valeurs scalaires aux requêtes complexes avec jointures et pagination.

Récupération de Valeurs Scalaires

La méthode scalar() est utile pour récupérer une seule valeur (par exemple, un compte, une somme ou un identifiant).

session_actuelle = SessionEnPortee()
try:
    # Insérer quelques articles si ce n'est pas déjà fait
    if session_actuelle.query(Article).count() == 0:
        session_actuelle.add_all([Article(titre_article=f"Article {i}", contenu_principal=f"Contenu {i}") for i in range(1, 5)])
        session_actuelle.commit()

    # Récupérer un titre spécifique
    premier_article_id = session_actuelle.query(Article.identifiant_article).order_by(Article.identifiant_article).first()
    if premier_article_id:
        titre_unique = session_actuelle.query(Article.titre_article).filter_by(identifiant_article=premier_article_id).scalar()
        print(f"\nTitre d'article unique (ID {premier_article_id}): {titre_unique}")

    # Compter le nombre total d'articles
    total_articles = session_actuelle.query(func.count(Article.identifiant_article)).scalar()
    print(f"Nombre total d'articles dans la DB: {total_articles}")

    # Autre façon de compter
    total_articles_alt = session_actuelle.query(func.count('*')).select_from(Article).scalar()
    print(f"Nombre total d'articles (méthode alternative): {total_articles_alt}")

except Exception as e:
    session_actuelle.rollback()
    print(f"Erreur lors des requêtes scalaires : {e}")
finally:
    SessionEnPortee.remove()

Récupération d'Enregistrements Spécifiques

Plusieurs méthodes permettent de récupérer des enregistrements individuels, avec des comportements différents en cas de multiples résultats ou d'absence de résultat.

session_actuelle = SessionEnPortee()
try:
    id_cible = session_actuelle.query(Article.identifiant_article).order_by(Article.identifiant_article).first()

    if id_cible:
        # first() : Retourne le premier résultat ou None si aucun.
        article_premier = session_actuelle.query(Article).filter_by(identifiant_article=id_cible).first()
        print(f"\nArticle (first) : {article_premier.titre_article}")

        # one() : Retourne le résultat s'il n'y en a qu'un, lève une exception sinon (trop ou pas assez de résultats).
        try:
            article_unique = session_actuelle.query(Article).filter_by(identifiant_article=id_cible).one()
            print(f"Article (one) : {article_unique.titre_article}")
        except Exception as e:
            print(f"Erreur avec .one() si plusieurs ou aucun résultat: {e}")

        # one_or_none() : Retourne le résultat s'il n'y en a qu'un, None si aucun, lève une exception si plusieurs.
        article_unique_ou_aucun = session_actuelle.query(Article).filter_by(identifiant_article=id_cible).one_or_none()
        if article_unique_ou_aucun:
            print(f"Article (one_or_none) : {article_unique_ou_aucun.titre_article}")

except Exception as e:
    session_actuelle.rollback()
    print(f"Erreur lors des requêtes d'enregistrements spécifiques : {e}")
finally:
    SessionEnPortee.remove()

Requêtes avec Jointures et Relations

Les relations définies avec relationship() gèrent les jointures de manière automatique. Le paramètre lazy contrôle comment et quand les données liées sont chargées.

• lazy='select' (par défaut) : Les données liées sont chargées lors du premier accès à l'attribut de relation, dans une requête SQL distincte. Fonctionne tant que la session est ouverte.
• lazy='joined' : Les données liées sont chargées via une jointure SQL (LEFT OUTER JOIN) dans la même requête que l'objet parent. Idéal pour les relations 1:1 ou 1:N lorsque les données liées sont presque toujours nécessaires.
• uselist=False : Indique qu'une relation 1:1 ou 1:0 est attendue, l'attribut de relation retournera un objet unique ou None, au lieu d'une liste.
• backref='nom_attribut_inverse' : Crée automatiquement une relation inverse sur le modèle lié.

session_actuelle = SessionEnPortee()
try:
    # Insérer un article avec propriétés pour les tests de relation
    article_relation = Article(
        titre_article="Article avec Propriétés",
        contenu_principal="Un article pour tester les relations.",
        details_article=ProprietesArticle(votes_total=50, vues_total=1000)
    )
    session_actuelle.add(article_relation)
    session_actuelle.commit()
    id_article_relation = article_relation.identifiant_article

    # Récupération paresseuse (lazy='select', par défaut)
    article_avec_lazy_select = session_actuelle.query(Article).filter_by(identifiant_article=id_article_relation).first()
    if article_avec_lazy_select:
        print(f"\nArticle (lazy='select'): {article_avec_lazy_select.titre_article}")
        # Les détails de l'article ne sont chargés qu'à ce moment-là
        if article_avec_lazy_select.details_article:
            print(f"  Détails (lazy loaded): Vues={article_avec_lazy_select.details_article.vues_total}")

    # Relation avec jointure (lazy='joined')
    # Pour illustrer, on peut temporairement modifier la relation ou en créer une nouvelle pour l'exemple
    class ArticleJoined(Article):
        __mapper_args__ = {
            'inherit_condition': Article.identifiant_article == Article.identifiant_article,
            'polymorphic_identity': 'article_joined'
        }
        details_article_joined = relationship('ProprietesArticle', lazy='joined', uselist=False, backref='article_complet')
    
    # Ou plus simplement, recharger l'objet avec joinedload
    from sqlalchemy.orm import joinedload
    article_avec_jointure = session_actuelle.query(Article).options(joinedload(Article.details_article)).filter_by(identifiant_article=id_article_relation).first()
    if article_avec_jointure and article_avec_jointure.details_article:
        print(f"Article (avec jointure explicite): {article_avec_jointure.titre_article}, Vues={article_avec_jointure.details_article.vues_total}")

    # Utilisation du backref
    # La relation ProprietesArticle.article_associe est créée par le backref dans Article.details_article
    prop_recuperees = session_actuelle.query(ProprietesArticle).filter_by(fk_article_id=id_article_relation).first()
    if prop_recuperees and prop_recuperees.article_associe:
        print(f"Accès inverse via backref (article_associe): {prop_recuperees.article_associe.titre_article}")

except Exception as e:
    session_actuelle.rollback()
    print(f"Erreur lors des requêtes avec relations : {e}")
finally:
    SessionEnPortee.remove()

Chargement Dynamique des Relations

Le paramètre lazy='dynamic' transforme la relation en un objet de requête, permettant d'appliquer des filtres, des limites ou des ordres sur les objets liés avant de les charger.

from sqlalchemy.sql import or_, and_

# Définition du modèle de commentaires
class CommentaireArticle(BaseDeclerative):
    __tablename__ = 'commentaires_articles'
    id_commentaire = Column(Integer, primary_key=True, autoincrement=True, comment='Clé primaire du commentaire')
    fk_article_id = Column(Integer, ForeignKey('articles.identifiant_article'), nullable=False, comment='Clé étrangère vers l\'article')
    contenu_commentaire = Column(String(5000), nullable=False)
    date_publication = Column(TIMESTAMP, server_default=text('CURRENT_TIMESTAMP'), comment='Date de publication du commentaire')

# Ajouter la relation de commentaires à la classe Article
Article.commentaires_lies = relationship('CommentaireArticle', lazy='dynamic', order_by=CommentaireArticle.date_publication.desc(), backref='article_du_commentaire')

# Re-créer les tables si nécessaire, ou utiliser des migrations
BaseDeclerative.metadata.create_all(moteur_db_app)

session_actuelle = SessionEnPortee()
try:
    # Insérer un article et quelques commentaires
    article_commentaires = Article(titre_article="Article avec Commentaires", contenu_principal="Contenu pour les commentaires.")
    session_actuelle.add(article_commentaires)
    session_actuelle.commit()
    id_article_commentaires = article_commentaires.identifiant_article

    for i in range(3):
        session_actuelle.add(CommentaireArticle(fk_article_id=id_article_commentaires, contenu_commentaire=f"Commentaire {i+1} pour l'article."))
    session_actuelle.commit()

    # Accès dynamique aux commentaires
    article_dyn = session_actuelle.query(Article).filter_by(identifiant_article=id_article_commentaires).first()
    if article_dyn:
        print(f"\nAccès dynamique aux commentaires de l'article '{article_dyn.titre_article}':")
        # commentaires_lies est un objet Query tant que .all(), .first(), etc. ne sont pas appelés
        requete_commentaires = article_dyn.commentaires_lies
        print(f"  Type d'objet de relation dynamique : {type(requete_commentaires)}")

        tous_commentaires = requete_commentaires.all()
        print(f"  Tous les commentaires : {[c.contenu_commentaire for c in tous_commentaires]}")

        commentaires_limites = requete_commentaires.limit(1).offset(0).all()
        print(f"  Commentaire paginé : {[c.contenu_commentaire for c in commentaires_limites]}")

except Exception as e:
    session_actuelle.rollback()
    print(f"Erreur lors du chargement dynamique : {e}")
finally:
    SessionEnPortee.remove()

Requêtes Conditionnelles Complexes

La méthode filter() est plus puissante que filter_by() car elle accepte des expressions SQL complètes et peut être combinée avec des opérateurs logiques comme and_ et or_.

session_actuelle = SessionEnPortee()
try:
    # Requêtes chaînées
    articles_filtres_chaines = session_actuelle.query(Article).filter(Article.identifiant_article > 0).filter(Article.date_creation > text("'2023-01-01'")).all()
    print(f"\nArticles (filtres chaînés): {[a.titre_article for a in articles_filtres_chaines]}")

    # Conditions multiples avec OR et AND
    conditions_complexes = or_(
        and_(Article.identifiant_article > 0, Article.identifiant_article < 100),
        and_(Article.identifiant_article >= 200, Article.identifiant_article < 300)
    )
    articles_par_conditions = session_actuelle.query(Article).filter(conditions_complexes).all()
    print(f"Articles (conditions complexes): {[a.titre_article for a in articles_par_conditions]}")

except Exception as e:
    session_actuelle.rollback()
    print(f"Erreur lors des requêtes complexes : {e}")
finally:
    SessionEnPortee.remove()

Pagination des Résultats

La pagination est essentielle pour les grands ensembles de résultats. SQLAlchemy fournit offset(), limit() et slice().

session_actuelle = SessionEnPortee()
try:
    # Pagination avec offset et limit
    articles_pagines_offset_limit = session_actuelle.query(Article).order_by(Article.date_creation.desc()).offset(0).limit(5).all()
    print(f"\nArticles paginés (offset 0, limit 5): {[a.titre_article for a in articles_pagines_offset_limit]}")

    # Pagination avec slice (équivalent à offset/limit mais avec indices Python)
    articles_pagines_slice = session_actuelle.query(Article).order_by(Article.date_creation.desc()).slice(0, 3).all()
    print(f"Articles paginés (slice 0, 3): {[a.titre_article for a in articles_pagines_slice]}")

except Exception as e:
    session_actuelle.rollback()
    print(f"Erreur lors de la pagination : {e}")
finally:
    SessionEnPortee.remove()

6. Gestion des Transactions

Les transactions sont cruciales pour maintenir l'intégrité et la cohérence des données. SQLAlchemy gère les transactions via la session.

session_tx = SessionEnPortee()
try:
    article_pour_tx = Article(titre_article="Article de Transaction", contenu_principal="Contenu destiné à une transaction.")
    session_tx.add(article_pour_tx)
    session_tx.flush() # Envoyer les modifications à la DB sans les commettre

    # Simuler une condition d'erreur
    if True: # Mettre False pour tester le commit réussi
        raise ValueError("Erreur simulée, la transaction sera annulée.")

    session_tx.commit() # Si aucune erreur, les changements sont sauvegardés
    print(f"\nTransaction réussie. Article '{article_pour_tx.titre_article}' ajouté avec l'ID {article_pour_tx.identifiant_article}.")

except ValueError as ve:
    session_tx.rollback() # Annuler toutes les opérations de la transaction en cas d'erreur
    print(f"\nTransaction annulée : {ve}")
except Exception as e:
    session_tx.rollback()
    print(f"\nUne erreur inattendue a annulé la transaction : {e}")
finally:
    SessionEnPortee.remove()

session.commit() rend toutes les modifications de la session permanentes. session.rollback() annule toutes les modifications non commises et restaure l'état précédent de la base de données. Il est essentiel d'utiliser ces méthodes conjointement dans un bloc try...except...finally pour garantir la robustesse des opérations.

7. Sessions Contextuelles

L'utilisation d'un gestionnaire de contexte (context manager) est une pratique élégante en Python pour s'assurer que les ressources (comme les sessions de base de données) sont correctement acquittées, même en cas d'erreur.

@contextlib.contextmanager
def obtenir_session_contextuelle(moteur_db):
    """
    Fournit une session SQLAlchemy gérée via un gestionnaire de contexte.
    La session est automatiquement commise ou rollbackée, puis fermée.
    """
    fabrique_locale_session = sessionmaker(bind=moteur_db)
    session_de_portee = scoped_session(fabrique_locale_session)
    session_instance = session_de_portee()
    try:
        yield session_instance
        session_instance.commit() # Valider les changements si tout va bien
    except Exception as ex:
        session_instance.rollback() # Annuler la transaction en cas d'erreur
        raise # Relaisser l'exception pour que le code appelant puisse la gérer
    finally:
        session_de_portee.remove() # Assurer que la session est retirée du contexte

# Utilisation du gestionnaire de contexte
with obtenir_session_contextuelle(moteur_db_app) as session_ctx:
    nombre_total = session_ctx.query(func.count(Article.identifiant_article)).scalar()
    print(f"\nNombre d'articles via session contextuelle : {nombre_total}")

# Exemple d'utilisation avec une écriture et une erreur simulée
try:
    with obtenir_session_contextuelle(moteur_db_app) as session_ctx_ecriture:
        nouvel_article_ctx = Article(titre_article="Article depuis contexte", contenu_principal="Contenu ajouté via un gestionnaire de contexte.")
        session_ctx_ecriture.add(nouvel_article_ctx)
        print(f"Article '{nouvel_article_ctx.titre_article}' ajouté (sera commis par le contexte).")

    with obtenir_session_contextuelle(moteur_db_app) as session_ctx_erreur:
        article_erreur_ctx = Article(titre_article="Article contextuel en erreur", contenu_principal="Ce contenu déclenchera un rollback.")
        session_ctx_erreur.add(article_erreur_ctx)
        raise ValueError("Erreur intentionnelle pour tester le rollback du contexte.")

except ValueError as e:
    print(f"Exception capturée : {e} (l'article de l'erreur a été annulé).")
except Exception as e:
    print(f"Une autre exception a été capturée : {e}.")