Introduction à SQLAlchemy

SQLAlchemy est un framework ORM (Object-Relational Mapping) implémenté en Python. Il permet de manipuler des bases de données relationnelles en utilisant des objets Python, facilitant ainsi la conversion entre les classes et les requêtes SQL. Contrairement à l'ORM intégré dans Django, SQLAlchemy est couramment utilisé avec des frameworks comme Flask ou FastAPI pour une gestion flexible des bases de données.

Installation

Pour installer SQLAlchemy, utilisez pip :

pip install sqlalchemy

Modélisation des tables

Pour définir des modèles de tables, commencez par importer les modules nécessaires et créer une classe de base :

from sqlalchemy import create_engine, Column, Integer, String, DateTime
from sqlalchemy.ext.declarative import declarative_base
import datetime

Base = declarative_base()

class Client(Base):
    __tablename__ = 'clients'
    identifiant = Column(Integer, primary_key=True)
    nom_complet = Column(String(100), nullable=False, index=True)
    adresse_email = Column(String(100), unique=True)
    date_creation = Column(DateTime, default=datetime.datetime.utcnow)

    __table_args__ = (
        {'UniqueConstraint': ('identifiant', 'nom_complet', name='uix_id_nom')},
        {'Index': ('ix_nom_email', 'nom_complet', 'adresse_email')},
    )

class Commande(Base):
    __tablename__ = 'commandes'
    id = Column(Integer, primary_key=True)
    reference = Column(String(50))

Le modèle Client représente une table avec des contraintes d'unicité et d'index. La classe Commande illustre un autre exemple de table.

Configuration du moteur de connexion

SQLAlchemy s'appuie sur des bibliothèques tierces comme pymysql pour se connecter aux bases de données. L'URL de connexion dépend du type de base de données :

# Connexion à MySQL via pymysql
engine = create_engine("mysql+pymysql://utilisateur:motdepasse@localhost:3306/nombd")

# Configuration avancée avec pool de connexions
engine = create_engine(
    "mysql+pymysql://utilisateur:motdepasse@localhost:3306/nombd",
    pool_size=5,
    max_overflow=2,
    pool_timeout=30,
    pool_recycle=3600
)

Pour sycnhroniser les modèles avec la base de données, exécutez le script suivant :

if __name__ == '__main__':
    Base.metadata.create_all(engine)
    # Base.metadata.drop_all(engine) pour supprimer toutes les tables

Gestion des sessions

Les sessions permettent de gérer les interactions avec la base de données. Créez une classe Session à partir du moteur :

from sqlalchemy.orm import sessionmaker

Session = sessionmaker(bind=engine)

def effectuer_operation():
    session = Session()
    nouveau_client = Client(nom_complet='Alice Martin', adresse_email='alice@example.com')
    session.add(nouveau_client)
    session.commit()
    session.close()

Pour une gestion thread-safe, utilisez scoped_session :

from sqlalchemy.orm import scoped_session, sessionmaker

session_locale = scoped_session(Session)

def tache_async():
    session_locale.add(Client(nom_complet='Bob Durand'))

Opérations CRUD

Création (Create)

client = Client(nom_complet='Charlie Dupont', adresse_email='charlie@example.com')
session.add(client)
session.commit()

# Ajout multiple
clients = [
    Client(nom_complet='David Lefebvre'),
    Client(nom_complet='Eva Moreau')
]
session.add_all(clients)
session.commit()

Lecture (Read)

# Sélection de tous les clients
tous_clients = session.query(Client).all()

# Filtrage par nom
client_specifique = session.query(Client).filter(Client.nom_complet == 'Alice Martin').first()

# Utilisation de filter_by pour des conditions simples
resultats = session.query(Client).filter_by(adresse_email='alice@example.com').all()

Mise à jour (Update)

# Modification via une requête
session.query(Client).filter(Client.identifiant == 1).update({'nom_complet': 'Alice Martin-Moreau'})
session.commit()

# Modification via un objet récupéré
client_a_modifier = session.query(Client).filter(Client.identifiant == 2).first()
client_a_modifier.adresse_email = 'nouvel_email@example.com'
session.add(client_a_modifier)
session.commit()

Suppression (Delete)

session.query(Client).filter(Client.identifiant == 3).delete()
session.commit()

Requêtes avancées

Projections et alias

from sqlalchemy.orm import aliased

resultat = session.query(Client.nom_complet.label('nom'), Client.adresse_email).all()

Requêtes avec filtres complexes

from sqlalchemy import and_, or_

# Filtrage avec plusieurs conditions
clients_filtrés = session.query(Client).filter(
    and_(Client.nom_complet.like('A%'), Client.identifiant > 10)
).all()

# Utilisation de OR
clients_alternatifs = session.query(Client).filter(
    or_(Client.adresse_email.endswith('@example.com'), Client.identifiant < 5)
).all()

Requêtes natives SQL

from sqlalchemy import text

requete = text("SELECT * FROM clients WHERE adresse_email LIKE :motif")
resultats = session.query(Client).from_statement(requete).params(motif='%@example.com').all()

Pagination et tri

clients_pagines = session.query(Client)[0:10]  # Les 10 premiers
clients_triés = session.query(Client).order_by(Client.date_creation.desc()).all()

Agrégations et groupement

from sqlalchemy.sql import func

statistiques = session.query(
    func.count(Client.identifiant).label('total'),
    func.min(Client.date_creation).label('premier_client')
).first()

# Groupement par mois
from sqlalchemy import extract

par_mois = session.query(
    extract('month', Client.date_creation).label('mois'),
    func.count('*').label('nombre')
).group_by('mois').all()

Relations entre tables

SQLAlchemy gère les relations une-à-plusieurs, une-à-une et plusieurs-à-plusieurs via des clés étrangères et l'attribut relationship.

from sqlalchemy.orm import relationship

class Departement(Base):
    __tablename__ = 'departements'
    id = Column(Integer, primary_key=True)
    nom = Column(String(50))
    employes = relationship('Employe', backref='departement')

class Employe(Base):
    __tablename__ = 'employes'
    id = Column(Integer, primary_key=True)
    nom = Column(String(50))
    dept_id = Column(Integer, ForeignKey('departements.id'))

class EmployeDetail(Base):
    __tablename__ = 'employe_details'
    id = Column(Integer, primary_key=True)
    adresse = Column(String(200))
    employe_id = Column(Integer, ForeignKey('employes.id'), unique=True)
    employe = relationship('Employe', backref='detail', uselist=False)

# Table associative pour plusieurs-à-plusieurs
ProjetEmploye = Table('projet_employe', Base.metadata,
    Column('projet_id', Integer, ForeignKey('projets.id')),
    Column('employe_id', Integer, ForeignKey('employes.id'))
)

class Projet(Base):
    __tablename__ = 'projets'
    id = Column(Integer, primary_key=True)
    titre = Column(String(100))
    participants = relationship('Employe', secondary=ProjetEmploye, backref='projets')

Utilisation des relations

# Création avec relations
dept = Departement(nom='Technologie')
employe = Employe(nom='Sophie Leroy', departement=dept)
session.add_all([dept, employe])
session.commit()

# Consultation des relations
employe_récupéré = session.query(Employe).filter_by(nom='Sophie Leroy').first()
print(employe_récupéré.departement.nom)  # Accès au département
print(employe_récupéré.detail.adresse)   # Accès aux détails (relation one-to-one)