Installation et Configuration

Pour intégrer SQLAlchemy dans votre environnemetn, exécutez la commande suivante :

pip install sqlalchemy

Si vous devez interagir avec un système de gestion de base de données spécifique, installez également le pilote correspondant :

# Pour PostgreSQL
pip install psycopg2-binary

# Pour MySQL
pip install pymysql

# SQLite est inclus nativement dans la bibliothèque standard de Python

Concepts Fondamentaux

• Moteur (Engine) : Gère le pool de connexions et assure la communication avec la base de données.
• Session : Espace de travail qui orchestre toutes les opérations de persistance des objets.
• Modèle (Model) : Représentation orientée objet des tables relationnelles.
• Requête (Query) : Interface permettant de construire et d'exécuter des instructions SQL de manière programmatique.

Établissement de la Connexion

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

# Initialisation du moteur de base de données (exemple avec SQLite)
db_engine = create_engine('sqlite:///corporate_data.db', echo=False)

# Configuration de la fabrique de sessions
SessionFactory = sessionmaker(bind=db_engine, autoflush=False, autocommit=False)

# Création d'une instance de session active
db_session = SessionFactory()

Modélsiation des Données

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

Base = declarative_base()

# Table d'association pour gérer la relation plusieurs-à-plusieurs
project_tech = Table(
    'project_tech_stack', Base.metadata,
    Column('project_id', Integer, ForeignKey('projects.id'), primary_key=True),
    Column('tech_id', Integer, ForeignKey('technologies.id'), primary_key=True)
)

class Employee(Base):
    __tablename__ = 'employees'
    
    id = Column(Integer, primary_key=True)
    full_name = Column(String(100), nullable=False)
    email = Column(String(150), unique=True)
    
    projects = relationship("Project", back_populates="lead")

class Project(Base):
    __tablename__ = 'projects'
    
    id = Column(Integer, primary_key=True)
    title = Column(String(200), nullable=False)
    description = Column(String(1000))
    lead_id = Column(Integer, ForeignKey('employees.id'))
    
    lead = relationship("Employee", back_populates="projects")
    technologies = relationship("Technology", secondary=project_tech, back_populates="projects")

class Technology(Base):
    __tablename__ = 'technologies'
    
    id = Column(Integer, primary_key=True)
    name = Column(String(50), unique=True, nullable=False)
    
    projects = relationship("Project", secondary=project_tech, back_populates="technologies")

Génération du Schéma

# Création des tables dans la base de données
Base.metadata.create_all(bind=db_engine)

Opérations CRUD de Base

Création

# Insertion d'un seul enregistrement
new_dev = Employee(full_name="Alice Martin", email="alice.martin@corp.com")
db_session.add(new_dev)
db_session.commit()

# Insertion multiple
db_session.add_all([
    Employee(full_name="Bob Chen", email="bob.chen@corp.com"),
    Employee(full_name="Charlie Doe", email="charlie.doe@corp.com")
])
db_session.commit()

Lecture

# Récupération de tous les employés
all_devs = db_session.query(Employee).all()

# Récupération du premier résultat
first_dev = db_session.query(Employee).first()

# Récupération par identifiant primaire
dev_by_id = db_session.query(Employee).get(1)

Mise à jour

# Modification d'un objet existant
target_dev = db_session.query(Employee).get(1)
target_dev.full_name = "Alice M. Martin"
db_session.commit()

# Mise à jour conditionnelle en masse
db_session.query(Employee).filter(Employee.email.like('%@corp.com')).update({"full_name": "Corp Employee"}, synchronize_session=False)
db_session.commit()

Suppression

# Suppression d'un objet spécifique
dev_to_remove = db_session.query(Employee).get(2)
db_session.delete(dev_to_remove)
db_session.commit()

# Suppression en masse
db_session.query(Employee).filter(Employee.full_name == "Bob Chen").delete(synchronize_session=False)
db_session.commit()

Requêtes Avancées

Requêtes de Base

# Tri et pagination
devs = db_session.query(Employee).order_by(Employee.full_name.desc()).limit(5).offset(2).all()

# Sélection de colonnes spécifiques
emails = db_session.query(Employee.email).all()

Filtrage

from sqlalchemy import or_

# Filtrage exact
exact_match = db_session.query(Employee).filter(Employee.full_name == "Alice M. Martin").first()

# Recherche approximative
like_match = db_session.query(Employee).filter(Employee.full_name.ilike("alice%")).all()

# Filtre IN
in_match = db_session.query(Employee).filter(Employee.id.in_([1, 3, 5])).all()

# Conditions multiples avec OR
complex_filter = db_session.query(Employee).filter(
    or_(Employee.full_name == "Alice M. Martin", Employee.email.like("%@corp.com"))
).all()

Agrégation

from sqlalchemy import func

# Comptage total
total_employees = db_session.query(func.count(Employee.id)).scalar()

# Dénombrement avec jointure et groupement
tech_count_per_project = db_session.query(
    Project.title, 
    func.count(Technology.id)
).join(project_tech).join(Technology).group_by(Project.title).all()

Jointures

# Jointure interne
projects_with_leads = db_session.query(Project, Employee).join(Employee).filter(Project.title.ilike("%AI%")).all()

# Jointure externe gauche
all_projects_with_leads = db_session.query(Project, Employee).outerjoin(Employee).all()

Gestion des Relations

# Création d'objets avec relations implicites
lead_dev = Employee(full_name="David Smith", email="david.smith@corp.com")
new_project = Project(title="Refactoring Core", description="Main system overhaul", lead=lead_dev)
db_session.add(new_project)
db_session.commit()

# Navigation à travers les relations
print(f"Le projet '{new_project.title}' est dirigé par {new_project.lead.full_name}")

# Manipulation de la relation plusieurs-à-plusieurs
python_tech = Technology(name="Python")
rust_tech = Technology(name="Rust")

new_project.technologies.append(python_tech)
new_project.technologies.append(rust_tech)
db_session.commit()

for tech in new_project.technologies:
    print(f"Technologie utilisée : {tech.name}")

Contrôle des Transactions

# Gestion manuelle avec rollback
try:
    temp_emp = Employee(full_name="Temp User", email="temp@corp.com")
    db_session.add(temp_emp)
    db_session.commit()
except Exception as err:
    db_session.rollback()
    print(f"Erreur de transaction : {err}")

# Utilisation de savepoints avec begin_nested
with db_session.begin_nested():
    nested_emp = Employee(full_name="Nested User", email="nested@corp.com")
    db_session.add(nested_emp)

Recommandations et Bonnes Pratiques

from contextlib import contextmanager

@contextmanager
def get_database_session():
    session = SessionFactory()
    try:
        yield session
        session.commit()
    except Exception:
        session.rollback()
        raise
    finally:
        session.close()

# Utilisation du gestionnaire de contexte
with get_database_session() as active_session:
    active_session.add(Employee(full_name="Context User", email="ctx@corp.com"))