SQLAlchemy - un ORM pour Python

Bienvenue ! La dernière fois, tu as écrit du SQL brut - cursor.execute("SELECT * FROM species"). Ça marche, mais chaque requête est une chaîne, et une faute de frappe dans un nom de colonne n'apparaît qu'à l'exécution. Découvre l'ORM (Object-Relational Mapping) - travailler avec une base de données via des classes Python plutôt qu'en SQL !

Analogie safari : Le SQL brut, c'est comme remplir des formulaires papier à la main. L'ORM SQLAlchemy est un système numérique - des objets Python mappés automatiquement vers les tables de la base !

Qu'est-ce qu'un ORM ?

Un ORM fait correspondre des objets aux tables d'une base : une classe est une table, un objet une ligne, un attribut une colonne. En SQL brut, enregistrer un lion, c'est une chaîne avec des points d'interrogation à la place des valeurs :

1cursor.execute("INSERT INTO species (name, population) VALUES (?, ?)", ("Lion", 120))

Avec un ORM, tu crées un objet ordinaire, et la session prépare elle-même l'INSERT :

1lion = Species(name="Lion", population=120)
2session.add(lion)
3session.commit()

Les deux ajoutent la même ligne, mais dans le second c'est SQLAlchemy qui écrit le SQL.

Les avantages de l'ORM :

• Des classes Python au lieu de chaînes SQL
• Type hints et autocomplétion
• Moins vulnérable à l'injection SQL
• Relations entre tables plus simples
• Migrations et évolution du schéma
• Indépendance vis-à-vis de la base (PostgreSQL, MySQL, SQLite...)

SQLAlchemy - l'ORM le plus populaire en Python

SQLAlchemy est un ORM puissant + un outil pour travailler avec les bases SQL. Nous utilisons la série actuelle 2.0.

Installation

Elle ne fait pas partie de Python standard, donc installe-la une fois depuis le terminal :

1pip install sqlalchemy

Pour SQLite, c'est tout - le pilote sqlite3 est intégré à Python.

Les bases de SQLAlchemy - les modèles

Un modèle est une classe qui décrit une table : son nom va dans __tablename__, ses colonnes sont des attributs Column avec types et contraintes. Cinq étapes mènent à une session prête :

1from sqlalchemy import create_engine, Column, Integer, String, Boolean
2from sqlalchemy.orm import declarative_base, sessionmaker
3
4# 1. Créer Base - la classe de base pour les modèles
5Base = declarative_base()
6
7# 2. Définir un modèle - une classe Python = une table SQL
8class Species(Base):
9    __tablename__ = 'species'  # Nom de la table
10
11    # Colonnes
12    id = Column(Integer, primary_key=True, autoincrement=True)
13    scientific_name = Column(String, nullable=False, unique=True)
14    common_name = Column(String, nullable=False)
15    population = Column(Integer, default=0)
16    habitat = Column(String)
17    endangered = Column(Boolean, default=False)
18
19    def __repr__(self):
20        return f"<Species('{self.common_name}', pop={self.population})>"
21
22
23# 3. Se connecter à la base de données
24engine = create_engine('sqlite:///safari_orm.db', echo=True)  # echo=True affiche le SQL
25
26# 4. Créer les tables
27Base.metadata.create_all(engine)
28
29# 5. Créer une session (gère les opérations sur la base)
30Session = sessionmaker(bind=engine)
31session = Session()

primary_key=True est le numéro unique de la ligne, nullable=False interdit les valeurs vides, unique=True les doublons. Pendant l'apprentissage, active echo=True pour voir le SQL généré. Remarque : Base.metadata.create_all() ne crée les tables que si elles n'existent pas (comme IF NOT EXISTS).

Les types de colonnes

MySQL exige une longueur de texte, par ex. String(100).

CRUD avec SQLAlchemy

CREATE - ajouter des enregistrements

Un nouvel enregistrement est un nouvel objet du modèle. Il entre dans la session et n'atteint la base qu'à la validation :

1# Créer un objet
2lion = Species(
3    scientific_name="Panthera leo",
4    common_name="Lion",
5    population=120,
6    habitat="savanna",
7    endangered=True
8)
9
10# Ajouter à la session
11session.add(lion)
12
13# Valider
14session.commit()
15
16print(f"Lion ajouté ID : {lion.id}")  # ID assigné automatiquement !
17
18# Ajouter plusieurs à la fois
19species_list = [
20    Species(scientific_name="Loxodonta africana", common_name="Elephant", population=450, endangered=True),
21    Species(scientific_name="Gorilla gorilla", common_name="Gorilla", population=230, endangered=True),
22]
23
24session.add_all(species_list)
25session.commit()

Avant commit(), le lion n'existe que dans la session et son id vaut None.

READ - récupérer des données

Une requête commence par session.query(Species), puis tu enchaînes les méthodes comme des maillons, en terminant par celle qui récupère les résultats :

1# Toutes les espèces
2all_species = session.query(Species).all()
3for species in all_species:
4    print(species.common_name, species.population)
5
6# Premier résultat
7first = session.query(Species).first()
8
9# Récupérer par ID
10lion = session.query(Species).get(1)  # ID = 1 (style legacy - avertissement en 2.0)
11# ou (SQLAlchemy 1.4+, recommandé)
12lion = session.get(Species, 1)
13
14# Filtrage - WHERE
15endangered = session.query(Species).filter(Species.endangered == True).all()
16savanna = session.query(Species).filter(Species.habitat == "savanna").all()
17
18# Plusieurs conditions (AND)
19results = session.query(Species).filter(
20    Species.endangered == True,
21    Species.population > 100
22).all()
23
24# OR
25from sqlalchemy import or_
26results = session.query(Species).filter(
27    or_(Species.habitat == "savanna", Species.habitat == "forest")
28).all()
29
30# LIKE
31results = session.query(Species).filter(Species.scientific_name.like("Panthera%")).all()
32
33# ORDER BY
34sorted_species = session.query(Species).order_by(Species.population.desc()).all()
35
36# LIMIT et OFFSET
37top_5 = session.query(Species).limit(5).all()
38page_2 = session.query(Species).limit(10).offset(10).all()
39
40# COUNT
41count = session.query(Species).count()
42endangered_count = session.query(Species).filter(Species.endangered == True).count()

filter() accepte les opérateurs Python, et SQLAlchemy les transforme en WHERE. Le style session.query() est « legacy » en 2.0 - le select() plus récent arrive dans le module FastAPI.

UPDATE - mettre à jour

Une mise à jour ORM, c'est juste un attribut modifié. La session le remarque et envoie une instruction UPDATE au commit() :

1# Méthode 1 : récupérer l'objet, modifier, valider
2lion = session.query(Species).filter(Species.common_name == "Lion").first()
3lion.population = 125
4session.commit()
5
6# Méthode 2 : mise à jour en masse
7session.query(Species).filter(Species.habitat == "savanna").update({
8    "endangered": True
9})
10session.commit()

La méthode 1 charge l'objet, la méthode 2 modifie de nombreuses lignes en une instruction, sans les charger.

DELETE - supprimer

Tu passes un objet unique à session.delete(), et plusieurs lignes sont supprimées par un filtre terminé par delete() :

1# Méthode 1 : récupérer l'objet, supprimer
2species_to_delete = session.get(Species, 10)
3if species_to_delete:
4    session.delete(species_to_delete)
5    session.commit()
6
7# Méthode 2 : suppression en masse
8session.query(Species).filter(Species.population == 0).delete()
9session.commit()

session.get() renvoie None quand l'enregistrement n'existe pas, d'où le if. Sans commit(), la modification est perdue.

Les relations - One-to-Many

Une espèce a de nombreuses observations de terrain : une clé étrangère les relie dans la base, relationship() en Python (exemple séparé et simplifié) :

1from sqlalchemy import ForeignKey
2from sqlalchemy.orm import relationship
3
4class Species(Base):
5    __tablename__ = 'species'
6
7    id = Column(Integer, primary_key=True)
8    common_name = Column(String, nullable=False)
9    population = Column(Integer, default=0)
10
11    # Relation : une espèce -> plusieurs observations
12    observations = relationship("Observation", back_populates="species", cascade="all, delete-orphan")
13
14
15class Observation(Base):
16    __tablename__ = 'observations'
17
18    id = Column(Integer, primary_key=True)
19    species_id = Column(Integer, ForeignKey('species.id'), nullable=False)
20    observation_date = Column(String, nullable=False)
21    location = Column(String, nullable=False)
22    count = Column(Integer, default=0)
23
24    # Relation : plusieurs observations -> une espèce
25    species = relationship("Species", back_populates="observations")

ForeignKey('species.id') pointe vers la colonne des espèces, back_populates relie les deux extrémités, et le réglage cascade supprime les observations avec leur espèce.

Ajoutons maintenant les observations du lion à sa liste et enregistrons tout dans la base avec un seul commit() :

1# Utilisation
2lion = Species(common_name="Lion", population=120)
3
4# Ajouter des observations au lion
5lion.observations.append(Observation(observation_date="2024-01-15", location="Serengeti", count=12))
6lion.observations.append(Observation(observation_date="2024-01-20", location="Masai Mara", count=8))
7
8session.add(lion)
9session.commit()
10
11# Récupérer les observations
12lion = session.query(Species).filter(Species.common_name == "Lion").first()
13for obs in lion.observations:
14    print(f"{obs.observation_date}: {obs.count}x à {obs.location}")

Nous n'avons jamais renseigné species_id - SQLAlchemy l'a rempli grâce à la relation.

Exemple safari - système ORM complet

Rassemblons tout dans une classe de service qui cache la session. D'abord, les modèles avec horodatage :

1from sqlalchemy import create_engine, Column, Integer, String, Boolean, ForeignKey, DateTime
2from sqlalchemy.orm import declarative_base, sessionmaker, relationship
3from datetime import datetime, timezone
4from typing import List, Optional
5
6Base = declarative_base()
7
8
9def utc_now() -> datetime:
10    return datetime.now(timezone.utc)
11
12
13class Species(Base):
14    """Modèle Species"""
15    __tablename__ = 'species'
16
17    id = Column(Integer, primary_key=True, autoincrement=True)
18    scientific_name = Column(String, nullable=False, unique=True)
19    common_name = Column(String, nullable=False)
20    population = Column(Integer, default=0)
21    habitat = Column(String)
22    endangered = Column(Boolean, default=False)
23    created_at = Column(DateTime, default=utc_now)
24    updated_at = Column(DateTime, default=utc_now, onupdate=utc_now)
25
26    # Relations
27    observations = relationship("Observation", back_populates="species", cascade="all, delete-orphan")
28
29    def __repr__(self):
30        return f"<Species('{self.common_name}', pop={self.population})>"
31
32
33class Observation(Base):
34    """Modèle Observation"""
35    __tablename__ = 'observations'
36
37    id = Column(Integer, primary_key=True, autoincrement=True)
38    species_id = Column(Integer, ForeignKey('species.id'), nullable=False)
39    observation_date = Column(String, nullable=False)
40    location = Column(String, nullable=False)
41    count = Column(Integer, default=0)
42    notes = Column(String)
43    created_at = Column(DateTime, default=utc_now)
44
45    # Relations
46    species = relationship("Species", back_populates="observations")
47
48    def __repr__(self):
49        return f"<Observation({self.species.common_name if self.species else 'N/A'}, {self.count}x @ {self.location})>"

default et onupdate reçoivent la fonction utc_now, pas son résultat, donc l'heure est calculée à chaque écriture. datetime.utcnow() est obsolète depuis Python 3.12.

La classe SafariORM crée le moteur et la session dans son constructeur, et chacune de ses méthodes effectue une opération CRUD :

1class SafariORM:
2    """Gestion de la base de données Safari via ORM"""
3
4    def __init__(self, db_url: str = "sqlite:///safari_orm.db"):
5        self.engine = create_engine(db_url, echo=False)
6        Base.metadata.create_all(self.engine)
7        Session = sessionmaker(bind=self.engine)
8        self.session = Session()
9
10    # === SPECIES ===
11
12    def create_species(self, scientific_name: str, common_name: str,
13                      population: int = 0, habitat: str = "",
14                      endangered: bool = False) -> Species:
15        """Ajouter une espèce"""
16        species = Species(
17            scientific_name=scientific_name,
18            common_name=common_name,
19            population=population,
20            habitat=habitat,
21            endangered=endangered
22        )
23        self.session.add(species)
24        self.session.commit()
25        return species
26
27    def get_species(self, species_id: int) -> Optional[Species]:
28        """Récupérer une espèce"""
29        return self.session.get(Species, species_id)
30
31    def list_species(self, endangered: Optional[bool] = None,
32                    habitat: Optional[str] = None) -> List[Species]:
33        """Lister les espèces"""
34        query = self.session.query(Species)
35
36        if endangered is not None:
37            query = query.filter(Species.endangered == endangered)
38        if habitat:
39            query = query.filter(Species.habitat == habitat)
40
41        return query.order_by(Species.common_name).all()
42
43    def update_species(self, species_id: int, **kwargs) -> Optional[Species]:
44        """Mettre à jour une espèce"""
45        species = self.session.get(Species, species_id)
46        if not species:
47            return None
48
49        for key, value in kwargs.items():
50            if hasattr(species, key):
51                setattr(species, key, value)
52
53        species.updated_at = utc_now()
54        self.session.commit()
55        return species
56
57    def delete_species(self, species_id: int) -> bool:
58        """Supprimer une espèce"""
59        species = self.session.get(Species, species_id)
60        if not species:
61            return False
62
63        self.session.delete(species)
64        self.session.commit()
65        return True
66
67    # === OBSERVATIONS ===
68
69    def create_observation(self, species_id: int, observation_date: str,
70                          location: str, count: int, notes: str = "") -> Optional[Observation]:
71        """Ajouter une observation"""
72        species = self.session.get(Species, species_id)
73        if not species:
74            return None
75
76        observation = Observation(
77            species_id=species_id,
78            observation_date=observation_date,
79            location=location,
80            count=count,
81            notes=notes
82        )
83        self.session.add(observation)
84        self.session.commit()
85        return observation
86
87    def get_observations_for_species(self, species_id: int) -> List[Observation]:
88        """Récupérer les observations d'une espèce"""
89        return self.session.query(Observation).filter(
90            Observation.species_id == species_id
91        ).order_by(Observation.observation_date.desc()).all()
92
93    def close(self):
94        """Fermer la session"""
95        self.session.close()

Les méthodes renvoient des modèles ou None, donc le reste du programme ne touche jamais au SQL ni à la session.

La démo utilise une base en mémoire, donc les exécutions répétées n'entrent jamais en conflit sur les noms uniques :

1# === DÉMONSTRATION ===
2
3print("=== SYSTÈME ORM SAFARI ===\n")
4
5db = SafariORM("sqlite:///:memory:")
6
7# 1. Ajouter des espèces
8print("1. Ajout des espèces (ORM)...")
9lion = db.create_species("Panthera leo", "Lion", 120, "savanna", True)
10elephant = db.create_species("Loxodonta africana", "Elephant", 450, "savanna", True)
11gorilla = db.create_species("Gorilla gorilla", "Gorilla", 230, "forest", True)
12
13print(f"   Ajoutés : {lion}, {elephant}, {gorilla}")
14
15# 2. Récupérer une espèce
16print("\n2. Récupération d'une espèce...")
17retrieved_lion = db.get_species(lion.id)
18print(f"   {retrieved_lion.common_name} : {retrieved_lion.population} individus")
19
20# 3. Lister les menacées
21print("\n3. Liste des espèces menacées...")
22endangered = db.list_species(endangered=True)
23for species in endangered:
24    print(f"   - {species.common_name} : {species.population}")
25
26# 4. Mise à jour
27print("\n4. Mise à jour de la population...")
28db.update_species(lion.id, population=125)
29lion = db.get_species(lion.id)
30print(f"   Nouvelle population du lion : {lion.population}")
31
32# 5. Ajouter des observations
33print("\n5. Ajout des observations...")
34db.create_observation(lion.id, "2024-01-15", "Serengeti", 12, "Pride with cubs")
35db.create_observation(lion.id, "2024-01-20", "Masai Mara", 8, "Male coalition")
36
37# 6. Récupérer les observations
38print("\n6. Observations du lion...")
39observations = db.get_observations_for_species(lion.id)
40for obs in observations:
41    print(f"   - {obs.observation_date} : {obs.count}x @ {obs.location}")
42
43# 7. Relations
44print("\n7. Navigation à travers les relations...")
45lion = db.get_species(lion.id)
46print(f"   Le lion a {len(lion.observations)} observations :")
47for obs in lion.observations:
48    print(f"     {obs.location} : {obs.count}x")
49
50db.close()
51print("\nDémonstration terminée")

L'adresse d'une base est toujours une URL comme sqlite:///fichier.db - un simple nom de fichier provoque une ArgumentError.

Un coup d'œil au prochain campement

Ensuite, Darwin te montre NoSQL MongoDB - une base de documents sans schéma rigide. Les mêmes étapes y ressemblent à ceci (un serveur MongoDB est nécessaire) :

1from pymongo import MongoClient
2
3client = MongoClient()      # connexion au serveur MongoDB
4db = client["safari"]       # choix de la base
5db.animals.insert_one({"name": "Lion", "population": 120})
6results = db.animals.find() # curseur avec les documents

Pas de modèle ni de create_all() - la collection animals apparaît à la première écriture.

Résumé

Dans cette leçon, tu as appris :

• Ce qu'est un ORM et à quoi il sert
• SQLAlchemy : Base, modèles, colonnes
• CRUD avec l'ORM : add, query, update, delete
• Le filtrage : filter(), order_by(), limit()
• Les relations : One-to-Many, relationship, ForeignKey
• Un système Safari complet avec ORM

Retiens : un ORM est un catalogue numérique - tu lui confies des objets Python, et les formulaires SQL se remplissent pour toi.