data / sqlalchemy

Requêtes : select, filter, order_by

Leçon 51 exercice

Explication

Ce que vous allez apprendre

Construire une requête avec select(Modele) puis l'exécuter via session.scalars(...)
Filtrer des résultats avec .where(...), and_, or_, not_
Trier des résultats avec .order_by(...) sur une ou plusieurs colonnes
Distinguer scalar_one() de scalar_one_or_none() selon le nombre de résultats attendu
Paginer une requête avec .limit() et .offset()

Dans quel contexte ?

Un développeur doit lister les utilisateurs actifs et majeurs d'une application, triés par nom, avec une pagination de 20 résultats par page. Plutôt que d'écrire cette requête en SQL brut, il construit un select(Utilisateur) étape par étape avec .where() et .order_by(), exactement comme le montre cette leçon, avant de l'exécuter une seule fois via la session.

D'abord, le problème à résoudre maintenant

Les leçons précédentes ont permis de créer, modifier et sauvegarder des objets. Mais une fois ces données en base, encore faut-il pouvoir les retrouver plus tard, sans écrire du SQL à la main à chaque fois.

Étape 1 : poser la question la plus simple

select(Utilisateur) pose la base de toute requête : "je veux des utilisateurs". À ce stade, rien n'est encore filtré ni trié, c'est juste une intention de départ, comme désigner le bon tiroir avant d'y chercher quelque chose de précis.

Étape 2 : ajouter une condition

.where(...) affine cette intention en ajoutant un filtre, par exemple "actifs et majeurs seulement". Chaque appel de méthode renvoie un NOUVEL objet requête, ce qui permet de la construire petit à petit, voire de rajouter des conditions seulement si certaines variables métier sont présentes.

Étape 3 : préciser l'ordre

.order_by(...) vient ensuite préciser dans quel ordre les résultats doivent sortir, exactement comme ORDER BY en SQL brut. Rien n'empêche de chaîner .where() puis .order_by() à la suite, la requête se construisant comme une phrase qu'on complète mot après mot.

Un point essentiel à bien distinguer : construire n'est pas exécuter

Écrire un select(...) ne déclenche AUCUNE requête vers la base : c'est une simple description, encore en mémoire côté Python. Il faut ensuite la confier à session.scalars(stmt).all() ou session.scalar(stmt) pour qu'elle parte réellement vers la base de données.

Pourquoi cette séparation existe

Ce découpage volontaire permet d'inspecter une requête, de la composer avec d'autres, ou de la réutiliser plusieurs fois avant de l'envoyer une seule fois. C'est ce qui rend possible la construction conditionnelle vue à l'étape 2.

Méthode d'exécution	Résultat	Comportement si 0 ou plusieurs lignes
`session.scalars(stmt).all()`	Liste de tous les résultats	Liste vide si aucun résultat
`session.scalar(stmt)`	Première valeur ou `None`	Silencieux
`.scalar_one()`	Une seule valeur garantie	Lève une exception si 0 ou plusieurs
`.scalar_one_or_none()`	Une valeur ou `None`	Lève une exception seulement si plusieurs

Un piège fréquent chez les débutants

.first() renvoie silencieusement le premier résultat, même s'il y en a plusieurs alors qu'on n'en attendait qu'un seul — un bug qui reste invisible longtemps. scalar_one() corrige ce problème : elle lève une exception si la requête renvoie zéro ou plusieurs lignes, ce qui force à détecter l'anomalie tout de suite plutôt qu'en production.

Piège fréquent

session.execute(select(Utilisateur).where(Utilisateur.email == email)).first() sur une colonne email qui n'est PAS déclarée unique=True peut silencieusement ignorer un doublon existant, alors que scalar_one() aurait immédiatement levé une exception claire signalant le problème de données.

Bonne pratique

Utilise scalar_one() ou scalar_one_or_none() dès que la logique métier attend exactement zéro/un résultat (recherche par identifiant unique ou email), et réserve .all()/scalars() aux listes où plusieurs résultats sont normaux.

Vers la suite

Ces mêmes principes de construction progressive d'une requête (select, where, order_by) s'appliquent aussi bien à une seule table qu'à plusieurs à la fois : la prochaine leçon les réutilise directement pour combiner des tables entre elles avec des jointures.

Commandes & code

Requêtes avec select()

from sqlalchemy import select, and_, or_, not_

# SQLAlchemy 2.0 : select() remplace query() (toujours disponible mais legacy)
with SessionLocal() as session:
    # Récupérer tous les résultats
    stmt = select(Utilisateur)
    utilisateurs = session.scalars(stmt).all()

    # Filtrer avec where()
    stmt = select(Utilisateur).where(Utilisateur.est_actif == True)
    actifs = session.scalars(stmt).all()

    # Plusieurs conditions (AND implicite en chaînant where(), explicite avec and_)
    stmt = select(Utilisateur).where(
        Utilisateur.est_actif == True,
        Utilisateur.age >= 18,
    )

    stmt = select(Utilisateur).where(
        and_(Utilisateur.age >= 18, Utilisateur.age < 65)
    )

    stmt = select(Utilisateur).where(
        or_(Utilisateur.nom == "Alice", Utilisateur.nom == "Bob")
    )

    stmt = select(Utilisateur).where(not_(Utilisateur.est_actif))

    # LIKE et IN
    stmt = select(Utilisateur).where(Utilisateur.email.like("%@entreprise.com"))
    stmt = select(Utilisateur).where(Utilisateur.id.in_([1, 2, 3]))

    # Une seule ligne
    utilisateur = session.scalar(select(Utilisateur).where(Utilisateur.email == "a@x.com"))
    utilisateur = session.execute(
        select(Utilisateur).where(Utilisateur.id == 1)
    ).scalar_one()            # lève une exception si 0 ou plusieurs résultats
    utilisateur = session.execute(
        select(Utilisateur).where(Utilisateur.id == 1)
    ).scalar_one_or_none()    # None si absent, exception si plusieurs

    # Tri
    stmt = select(Utilisateur).order_by(Utilisateur.nom.asc())
    stmt = select(Utilisateur).order_by(Utilisateur.cree_le.desc())
    stmt = select(Utilisateur).order_by(Utilisateur.nom, Utilisateur.age.desc())

    # Pagination
    stmt = select(Utilisateur).order_by(Utilisateur.id).limit(20).offset(40)

    # Comptage
    from sqlalchemy import func
    total = session.scalar(select(func.count()).select_from(Utilisateur))

    # get() : raccourci le plus rapide pour une recherche par clé primaire
    u = session.get(Utilisateur, 1)   # utilise l'identity map avant d'aller en base

    # Sélectionner des colonnes précises (pas l'objet entier)
    stmt = select(Utilisateur.id, Utilisateur.email).where(Utilisateur.est_actif == True)
    for id_, email in session.execute(stmt):
        print(id_, email)

    # exists() pour vérifier une présence sans charger les données
    from sqlalchemy import exists
    a_des_commandes = session.scalar(
        select(exists().where(Commande.client_id == 1))
    )

Résumé

select(Modele) construit la requête, session.scalars(stmt).all() exécute et retourne les objets.
scalar_one()/scalar_one_or_none() imposent une cardinalité stricte (utile pour des lookups uniques).
session.get(Modele, pk) passe d'abord par l'identity map avant d'interroger la base.
Sélectionner des colonnes précises (select(Modele.id, Modele.email)) évite de charger l'objet complet quand inutile.

Exercices pratiques

1 disponible

Mission : un lookup par email qui masque un doublon de données

Objectif : Diagnostiquer pourquoi .first() masque un problème de données, puis corriger la requête pour qu'elle échoue bruyamment en cas de doublon.

Contexte

Le code d'authentification fait session.execute(select(Utilisateur).where(Utilisateur.email == email)).first(). La colonne email n'est PAS déclarée unique=True en base. Un incident récent a révélé qu'un même email existait en double, et que le mauvais compte était systématiquement retourné à la connexion, sans aucune erreur visible.

Résoudre l’exercice →