Table des matières

1. Installation de SQLAlchemy
2. Concepts clés
3. Connexion à la base de données
4. Définition des modèles de données
5. Création des tables
6. Opérations CRUD de base
7. Requêtes avancées
8. Opérations relationnelles
9. Gestion des transactions
10. Bonnes pratiques

Installation

bash

pip install sqlalchemy

Pour une base de données spécifique, installez le pilote approprié :

bash

# PostgreSQL
pip install psycopg2-binary

# MySQL
pip install mysql-connector-python

# SQLite (déjà inclus dans la bibliothèque standard de Python)

Concepts clés

• Engine : moteur de connexion à la base de données, responsable de la communication
• Session : session de base de données, gère toutes les opérations persistantes
• Model : classe de modèle de données, correspond à une table
• Query : objet de requête, utilisé pour construire et exécuter des requêtes

Connexion à la base de données

python

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

# Création du moteur de connexion
# Exemple SQLite
engine = create_engine('sqlite:///exemple.db', echo=True)

# Exemple PostgreSQL
# engine = create_engine('postgresql://utilisateur:motdepasse@localhost:5432/mabasededonnees')

# Exemple MySQL
# engine = create_engine('mysql+mysqlconnector://utilisateur:motdepasse@localhost:3306/mabasededonnees')

# Fabrique de sessions
SessionLocale = sessionmaker(autocommit=False, autoflush=False, bind=engine)

# Création d'une instance de session
session = SessionLocale()

Définition des modèles de données

python

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

# Création de la classe de base
Base = declarative_base()

class Utilisateur(Base):
    __tablename__ = 'utilisateurs'
    
    id = Column(Integer, primary_key=True, index=True)
    nom = Column(String(50), nullable=False)
    email = Column(String(100), unique=True, index=True)
    
    # Relation un-à-plusieurs
    articles = relationship("Article", back_populates="auteur")
    
class Article(Base):
    __tablename__ = 'articles'
    
    id = Column(Integer, primary_key=True, index=True)
    titre = Column(String(100), nullable=False)
    contenu = Column(String(500))
    auteur_id = Column(Integer, ForeignKey('utilisateurs.id'))
    
    # Relation plusieurs-à-un
    auteur = relationship("Utilisateur", back_populates="articles")
    
    # Relation plusieurs-à-plusieurs (via table d'association)
    etiquettes = relationship("Etiquette", secondary="articles_etiquettes", back_populates="articles")

class Etiquette(Base):
    __tablename__ = 'etiquettes'
    
    id = Column(Integer, primary_key=True, index=True)
    nom = Column(String(30), unique=True, nullable=False)
    
    articles = relationship("Article", secondary="articles_etiquettes", back_populates="etiquettes")

# Table d'association pour relation plusieurs-à-plusieurs
class ArticleEtiquette(Base):
    __tablename__ = 'articles_etiquettes'
    
    article_id = Column(Integer, ForeignKey('articles.id'), primary_key=True)
    etiquette_id = Column(Integer, ForeignKey('etiquettes.id'), primary_key=True)

Création des tables

python

# Création de toutes les tables
Base.metadata.create_all(bind=engine)

# Suppression de toutes les tables
# Base.metadata.drop_all(bind=engine)

Opérations CRUD de base

Création

python

# Création d'un nouvel utilisateur
nouvel_utilisateur = Utilisateur(nom="Jean Dupont", email="jean.dupont@exemple.com")
session.add(nouvel_utilisateur)
session.commit()

# Création en lots
session.add_all([
    Utilisateur(nom="Marie Curie", email="marie.curie@exemple.com"),
    Utilisateur(nom="Albert Einstein", email="albert.einstein@exemple.com")
])
session.commit()

Lecture

python

# Récupération de tous les utilisateurs
utilisateurs = session.query(Utilisateur).all()

# Premier utilisateur
premier = session.query(Utilisateur).first()

# Récupération par ID
utilisateur = session.query(Utilisateur).get(1)

Mise à jour

python

# Recherche puis mise à jour
utilisateur = session.query(Utilisateur).get(1)
utilisateur.nom = "Jean Dupont Modifié"
session.commit()

# Mise à jour en lots
session.query(Utilisateur).filter(Utilisateur.nom.like("Jean%")).update({"nom": "Jean"}, synchronize_session=False)
session.commit()

Suppression

python

# Recherche puis suppression
utilisateur = session.query(Utilisateur).get(1)
session.delete(utilisateur)
session.commit()

# Suppression en lots
session.query(Utilisateur).filter(Utilisateur.nom == "Marie Curie").delete(synchronize_session=False)
session.commit()

Requêtes avancées

Requêtes de base

python

# Tous les enregistrements
utilisateurs = session.query(Utilisateur).all()

# Champs spécifiques
noms = session.query(Utilisateur.nom).all()

# Tri
utilisateurs = session.query(Utilisateur).order_by(Utilisateur.nom.desc()).all()

# Limitation
utilisateurs = session.query(Utilisateur).limit(10).all()

# Décalage
utilisateurs = session.query(Utilisateur).offset(5).limit(10).all()

Filtres

python

from sqlalchemy import or_

# Filtre égalité
utilisateur = session.query(Utilisateur).filter(Utilisateur.nom == "Jean Dupont").first()

# Filtre flou
utilisateurs = session.query(Utilisateur).filter(Utilisateur.nom.like("Jean%")).all()

# Filtre IN
utilisateurs = session.query(Utilisateur).filter(Utilisateur.nom.in_(["Jean Dupont", "Marie Curie"])).all()

# Conditions multiples
utilisateurs = session.query(Utilisateur).filter(
    Utilisateur.nom == "Jean Dupont",
    Utilisateur.email.like("%@exemple.com")
).all()

# Condition OU
utilisateurs = session.query(Utilisateur).filter(
    or_(Utilisateur.nom == "Jean Dupont", Utilisateur.nom == "Marie Curie")
).all()

# Différent de
utilisateurs = session.query(Utilisateur).filter(Utilisateur.nom != "Jean Dupont").all()

Requêtes d'agrégation

python

from sqlalchemy import func

# Comptage
total = session.query(Utilisateur).count()

# Comptage par groupe
comptes_articles = session.query(
    Utilisateur.nom,
    func.count(Article.id)
).join(Article).group_by(Utilisateur.nom).all()

# Moyenne, somme, etc.
moyenne_id = session.query(func.avg(Utilisateur.id)).scalar()

Requêtes de jointure

python

# Jointure interne
resultats = session.query(Utilisateur, Article).join(Article).filter(Article.titre.like("%Python%")).all()

# Jointure externe gauche
resultats = session.query(Utilisateur, Article).outerjoin(Article).all()

# Condition de jointure explicite
resultats = session.query(Utilisateur, Article).join(Article, Utilisateur.id == Article.auteur_id).all()

Opérations relationnelles

python

# Création d'objets liés
utilisateur = Utilisateur(nom="Sophie Martin", email="sophie.martin@exemple.com")
article = Article(titre="Mon premier blog", contenu="Bonjour tout le monde !", auteur=utilisateur)
session.add(article)
session.commit()

# Accès via relations
print(f"L'auteur de l'article '{article.titre}' est {article.auteur.nom}")
print(f"Tous les articles de {utilisateur.nom} :")
for art in utilisateur.articles:
    print(f"  - {art.titre}")

# Opération plusieurs-à-plusieurs
python_etiquette = Etiquette(nom="Python")
sqlalchemy_etiquette = Etiquette(nom="SQLAlchemy")

article.etiquettes.append(python_etiquette)
article.etiquettes.append(sqlalchemy_etiquette)
session.commit()

print(f"Étiquettes de l'article '{article.titre}' :")
for tag in article.etiquettes:
    print(f"  - {tag.nom}")

Gestion des transactions

python

# Validation automatique avec rollback
try:
    utilisateur = Utilisateur(nom="Test", email="test@exemple.com")
    session.add(utilisateur)
    session.commit()
except Exception as e:
    session.rollback()
    print(f"Erreur : {e}")

# Utilisation d'un gestionnaire de contexte transactionnel
from sqlalchemy.orm import Session

def creer_utilisateur(session: Session, nom: str, email: str):
    try:
        utilisateur = Utilisateur(nom=nom, email=email)
        session.add(utilisateur)
        session.commit()
        return utilisateur
    except:
        session.rollback()
        raise

# Transaction imbriquée
with session.begin_nested():
    utilisateur = Utilisateur(nom="Transaction", email="transaction@exemple.com")
    session.add(utilisateur)

# Point de sauvegarde
savepoint = session.begin_nested()
try:
    utilisateur = Utilisateur(nom="Point de sauvegarde", email="savepoint@exemple.com")
    session.add(utilisateur)
    savepoint.commit()
except:
    savepoint.rollback()

Bonnes pratiques

1. Gestion des sessions : créez une nouvelle session par requête et fermez-la après usage
2. Gestion des exceptions : gérez toujours les exceptions et effectuez un rollback approprié
3. Chargement paresseux : attention au problème N+1, optimisez avec le chargement immédiat (eager loading)
4. Pool de connexions : configurez correctement la taille du pool et les délais d'attente
5. Validation des données : validez l'intégrité des données au niveau du modèle ou de l'application

python

# Gestion des sessions avec un gestionnaire de contexte
from contextlib import contextmanager

@contextmanager
def obtenir_db():
    db = SessionLocale()
    try:
        yield db
        db.commit()
    except Exception:
        db.rollback()
        raise
    finally:
        db.close()

# Exemple d'utilisation
with obtenir_db() as db:
    utilisateur = Utilisateur(nom="Contexte", email="contexte@exemple.com")
    db.add(utilisateur)

Résumé

SQLAlchemy ORM fournit un moyen puissant et flexible d'interagir avec les bases de données. Après avoir lu cet article, vous devriez être en mesure de :

1. Installer et configruer SQLAlchemy
2. Définir des modèles de données et des relations
3. Effectuer des opérations CRUD de base
4. Construire des requêtes complexes
5. Gérer les transactions
6. Appliquer les bonnes pratiques

SQLAlchemy offre encore plus de fonctionnalités avancées comme les propriétés hybrides, les écouteurs d'événements, les requêtes personnalisées, qui valent la peine d'être explorées.