Utiliser SQLAlchemy ORM pour les opérations sur la base de données en Python

SQLAlchemy est l'un des frameworks ORM (Object-Relational Mapping) les plus prisés en Python, offrant une approche efficace et flexible pour interagir avec les bases de données. Ce guide détaille comment employer SQLAlchemy ORM pour réaliser diverses opérations sur votre base de données.

Installation

Pour installer SQLAlchemy, exécutez la commande suivante :

pip install sqlalchemy

Si vous prévoyez de vous connecter à une base de données spécifique, vous devrez également installer le pilote correspondant :

# Pour PostgreSQL
pip install psycopg2-binary

# Pour MySQL
pip install mysql-connector-python

# Pour SQLite (inclus dans la bibliothèque standard de Python, pas d'installation supplémentaire nécessaire)

Concepts Clés

  • Engine : Le moteur de connnexion à la base de données, responsable de la communication.
  • Session : La session de base de données, qui gère toutes les opérations de persistence.
  • Model : Les classes représentant les modèles de données, qui correspondent aux tables de la base de données.
  • Query : Les objets de requête, utilisés pour construire et exécuter des requêtes sur la base de données.

Connexion à la Base de Données

Établissez une connexion à votre base de données en utilisant l'objet Engine.

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

# Création de l'engine de connexion. Exemple avec SQLite :
engine = create_engine('sqlite:///mydatabase.db', echo=True)

# Exemple avec PostgreSQL :
# engine = create_engine('postgresql://user:password@host:port/database_name')

# Exemple avec MySQL :
# engine = create_engine('mysql+mysqlconnector://user:password@host:port/database_name')

# Configuration de la fabrique de sessions
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

# Instanciation d'une session
session = SessionLocal()

Définition des Modèles de Données

Définissez vos modèles de données sous forme de classes Python, héritant de declarative_base.

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

Base = declarative_base()

class Author(Base):
    __tablename__ = 'authors'
    
    author_id = Column(Integer, primary_key=True, index=True)
    full_name = Column(String(100), nullable=False)
    email_address = Column(String(150), unique=True, index=True)
    
    # Relation One-to-Many avec les articles
    articles = relationship("Article", back_populates="author")
    
class Article(Base):
    __tablename__ = 'articles'
    
    article_id = Column(Integer, primary_key=True, index=True)
    article_title = Column(String(200), nullable=False)
    content_body = Column(String(1000))
    author_ref_id = Column(Integer, ForeignKey('authors.author_id'))
    
    # Relation Many-to-One avec l'auteur
    author = relationship("Author", back_populates="articles")
    
    # Relation Many-to-Many avec les étiquettes (via une table de liaison)
    tags = relationship("Tag", secondary="article_tags", back_populates="articles")

class Tag(Base):
    __tablename__ = 'tags'
    
    tag_id = Column(Integer, primary_key=True, index=True)
    tag_name = Column(String(50), unique=True, nullable=False)
    
    articles = relationship("Article", secondary="article_tags", back_populates="tags")

# Table de liaison pour la relation Many-to-Many
class ArticleTag(Base):
    __tablename__ = 'article_tags'
    
    article_link_id = Column(Integer, ForeignKey('articles.article_id'), primary_key=True)
    tag_link_id = Column(Integer, ForeignKey('tags.tag_id'), primary_key=True)

Création des Tables

Pour créer les tables dans votre base de données basées sur les modèles définis :

# Crée toutes les tables définies dans metadata
Base.metadata.create_all(bind=engine)

# Pour supprimer toutes les tables (à utiliser avec prudence) :
# Base.metadata.drop_all(bind=engine)

Opérations CRUD de Base

Création d'enregistrements

Ajoutez de nouveaux enregistrements dans vos tables.

# Créer un nouvel auteur
new_author = Author(full_name="Alice Dupont", email_address="alice.dupont@email.com")
session.add(new_author)
session.commit()

# Création en masse
session.add_all([
    Author(full_name="Bob Martin", email_address="bob.martin@email.com"),
    Author(full_name="Charlie Brown", email_address="charlie.brown@email.com")
])
session.commit()

Lecture d'enregistrements

Récupérez des données depuis la base de données.

# Obtenir tous les auteurs
all_authors = session.query(Author).all()

# Obtenir le premier auteur trouvé
first_author = session.query(Author).first()

# Obtenir un auteur par son ID
author_by_id = session.query(Author).get(1)

Mise à jour d'enregistrements

Modifiez des enregistrements existants.

# Récupérer un auteur et le modifier
author_to_update = session.query(Author).get(1)
if author_to_update:
    author_to_update.full_name = "Alice Dubois"
    session.commit()

# Mise à jour de plusieurs enregistrements
session.query(Author).filter(Author.full_name.like("A%")).update({"full_name": "Mme A."}, synchronize_session=False)
session.commit()

Suppression d'enregistrements

Supprimez des enregistrements de la base de données.

# Récupérer un auteur et le supprimer
author_to_delete = session.query(Author).get(1)
if author_to_delete:
    session.delete(author_to_delete)
    session.commit()

# Suppression en masse
session.query(Author).filter(Author.full_name == "Bob Martin").delete(synchronize_session=False)
session.commit()

Interrogation de Données

Requêtes de base

Effectuez des requêtes simples pour filltrer et trier les données.

# Sélectionner uniquement les noms des auteurs
author_names = session.query(Author.full_name).all()

# Trier les auteurs par nom
sorted_authors = session.query(Author).order_by(Author.full_name.desc()).all()

# Limiter le nombre de résultats
limited_authors = session.query(Author).limit(5).all()

# Appliquer un offset
paginated_authors = session.query(Author).offset(10).limit(5).all()

Filtrage de requêtes

Appliquez des conditions de filtrage complexes.

from sqlalchemy import or_

# Filtrage par égalité
specific_author = session.query(Author).filter(Author.full_name == "Charlie Brown").first()

# Recherche par motif (LIKE)
authors_starting_with_C = session.query(Author).filter(Author.full_name.like("C%")).all()

# Filtrage IN
authors_in_list = session.query(Author).filter(Author.full_name.in_(["Alice Dupont", "Bob Martin"])).all()

# Combinaison de conditions
filtered_authors = session.query(Author).filter(
    Author.full_name == "Alice Dupont", 
    Author.email_address.endswith("@email.com")
).all()

# Condition OR
authors_or_condition = session.query(Author).filter(
    or_(Author.full_name == "Alice Dupont", Author.full_name == "Bob Martin")
).all()

# Condition de non-égalité
authors_not_equal = session.query(Author).filter(Author.full_name != "Charlie Brown").all()

Requêtes d'agrégation

Utilisez des fonctions d'agrégation comme COUNT, AVG, etc.

from sqlalchemy import func

# Compter le nombre total d'auteurs
total_authors = session.query(Author).count()

# Compter le nombre d'articles par auteur
author_article_counts = session.query(
    Author.full_name, 
    func.count(Article.article_id)
).join(Article).group_by(Author.full_name).all()

# Calculer la moyenne d'un champ
average_author_id = session.query(func.avg(Author.author_id)).scalar()

Requêtes avec jointures

Combinez des données de plusieurs tables.

# Jointure interne (INNER JOIN)
author_articles = session.query(Author, Article).join(Article).filter(Article.article_title.like("%SQL%")).all()

# Jointure externe gauche (LEFT OUTER JOIN)
all_authors_with_articles = session.query(Author, Article).outerjoin(Article).all()

# Jointure avec une condition spécifique
explicit_join = session.query(Author, Article).join(Article, Author.author_id == Article.author_ref_id).all()

Gestion des Relations

Interagissez avec les objets liés.

# Créer un article lié à un auteur
new_author = Author(full_name="David Lee", email_address="david.lee@email.com")
new_article = Article(article_title="Introduction à SQLAlchemy", content_body="Détails...", author=new_author)
session.add(new_article)
session.commit()

# Accéder aux objets liés
print(f"L'auteur de l'article '{new_article.article_title}' est {new_article.author.full_name}")
print(f"Articles de {new_author.full_name} :")
for art in new_author.articles:
    print(f"  - {art.article_title}")

# Gestion des relations Many-to-Many
python_tag = Tag(tag_name="Python")
sqlalchemy_tag = Tag(tag_name="SQLAlchemy")

new_article.tags.append(python_tag)
new_article.tags.append(sqlalchemy_tag)
session.commit()

print(f"Étiquettes pour l'article '{new_article.article_title}' :")
for tag in new_article.tags:
    print(f"  - {tag.tag_name}")

Gestion des Transactions

Contrôlez les transactions pour garantir l'intégrité des données.

# Gestion automatique des transactions avec commit/rollback
try:
    user_record = Author(full_name="Transaction User", email_address="trans@example.com")
    session.add(user_record)
    session.commit()
except Exception as e:
    session.rollback()
    print(f"Erreur lors de la transaction : {e}")

# Utilisation d'un gestionnaire de contexte pour les sessions
from sqlalchemy.orm import Session

def create_new_author(db_session: Session, name: str, email: str):
    try:
        new_author_rec = Author(full_name=name, email_address=email)
        db_session.add(new_author_rec)
        db_session.commit()
        return new_author_rec
    except:
        db_session.rollback()
        raise

# Transactions imbriquées avec savepoints
savepoint = session.begin_nested()
try:
    author1 = Author(full_name="Savepoint Author 1", email_address="sp1@example.com")
    session.add(author1)
    # Un autre bloc transactionnel imbriqué
    nested_savepoint = session.begin_nested()
    try:
        author2 = Author(full_name="Savepoint Author 2", email_address="sp2@example.com")
        session.add(author2)
        # Si tout va bien dans le bloc interne, on valide seulement ce bloc
        nested_savepoint.commit() 
    except:
        # Si une erreur survient dans le bloc interne, on annule seulement ce bloc
        nested_savepoint.rollback()
        raise
    savepoint.commit() # Valide le bloc externe si le bloc interne a réussi
except:
    savepoint.rollback() # Annule le bloc externe en cas d'erreur

Bonnes Pratiques

  • Gestion des Sessions : Créez une nouvelle session pour chaque requête web et fermez-la après utilisation.
  • Gestion des Erreurs : Implémentez une gestion robuste des exceptions avec des rollback appropriés.
  • Chargement Différé (Lazy Loading) : Soyez conscient du problème des requêtes N+1 et utilisez le chargement anticipé (eager loading) si nécessaire.
  • Pool de Connexions : Configurez judicieusement la taille du pool de connexions et les délais d'attente.
  • Validation des Données : Validez l'intégrité des données au niveau du modèle ou de l'application.

Voici un exemple d'utilisation d'un gestionnaire de contexte pour la gestion des sessions :

from contextlib import contextmanager

@contextmanager
def manage_db_session():
    db = SessionLocal()
    try:
        yield db
        db.commit() # Commit automatique si aucune exception n'est levée
    except Exception:
        db.rollback() # Rollback en cas d'exception
        raise
    finally:
        db.close() # Fermeture de la session

# Utilisation du gestionnaire de contexte
with manage_db_session() as current_db:
    new_context_author = Author(full_name="Context User", email_address="ctx@example.com")
    current_db.add(new_context_author)

Étiquettes: SQLAlchemy Python ORM Bases de données programmation

Publié le 6 juillet à 05h13

Étiquettes Populaires

©2026 L'Atelier Monstre