Un CRUD complet depuis Python : un module db.py complet et le test des quatre actions

Objectifs de la leçon

Assembler un module db.py complet avec toutes les actions CRUD sur une seule entité
Expliquer pour chaque fonction ce qu’elle renvoie dans chaque état — un dict, une liste, None, un bool ou une ValueError
Tester les quatre actions et leurs cas limites dans un seul script
Être prêt à relier le module à un serveur FastAPI

Qu’allons-nous apprendre ?

À la fin de cette leçon, comme le définit le programme, nous avons un CRUD complet depuis Python. La leçon assemble toutes les parties des leçons précédentes en un seul module, db.py, et le teste de bout en bout.

CRUD	Fonction	SQL	Renvoie
(préparation)	`init_db()`	`CREATE TABLE IF NOT EXISTS`	—
Create	`create_student(name, age, email)`	`INSERT`	Un dict (ou une `ValueError` sur un e-mail en double)
Read (tous)	`get_all_students()`	`SELECT`	Une liste de dicts (peut-être vide)
Read (un)	`get_student(student_id)`	`SELECT ... WHERE id = %s`	Un dict ou `None`
Update	`update_student(student_id, name, age, email)`	`UPDATE ... WHERE id = %s`	Un dict, `None`, ou une `ValueError`
Delete	`delete_student(student_id)`	`DELETE ... WHERE id = %s`	`True` / `False`

Pourquoi en avons-nous besoin ?

Ensuite, nous relierons ce module à un serveur. Chaque endpoint fera quelques lignes, qui appellent une fonction d’ici et traduisent le résultat en HTTP. Pour que cela fonctionne sans surprises, le module doit être complet, cohérent et testé, avec un « contrat » clair : ce que renvoie chaque fonction dans chaque état.

Explication à partir des bases

Le « contrat » d’une fonction

Contrat = la promesse de la fonction à celui qui l’appelle : ce qu’elle reçoit, ce qu’elle renvoie, et quelles erreurs elle peut lancer. Le tableau ci-dessus est le contrat de db.py. Le serveur s’appuiera sur lui :

get_student(...)    returned None     →  404
create_student(...) raised ValueError  →  409
delete_student(...) returned False    →  404

Les principes communs à toutes les fonctions

Une connexion pour chaque action — get_connection(), et close() dans finally.
Des paramètres uniquement — %s, jamais de concaténation.
commit après chaque modification.
Un dict en sortie — cursor(dictionary=True).
« Introuvable » = None / False, et non une exception.
Une violation de contrainte = une ValueError avec un message clair.
Pas de HTTP, pas de print — le module ne sait pas qui l’utilise (un serveur, un script, un menu).

Syntaxe

La structure du module :

db.py
├── get_connection()            ← one place that opens connections (host, user, password, database)
├── init_db()                   ← create the table if needed
├── get_all_students()          ← R (list)
├── get_student(student_id)     ← R (one)
├── create_student(...)         ← C
├── update_student(...)         ← U
└── delete_student(student_id)  ← D

Premier exemple de code

Le db.py complet (dans le projet week-4/school-db) :

import mysql.connector

def get_connection():
    return mysql.connector.connect(
        host="localhost", user="root", password="kodkod", database="school")

def init_db():
    connection = get_connection()
    try:
        cursor = connection.cursor()
        cursor.execute(
            """
            CREATE TABLE IF NOT EXISTS students (
                id INT AUTO_INCREMENT PRIMARY KEY,
                name VARCHAR(100) NOT NULL,
                age INT NOT NULL,
                email VARCHAR(100) NOT NULL UNIQUE
            )
            """
        )
    finally:
        connection.close()

def get_all_students():
    connection = get_connection()
    try:
        cursor = connection.cursor(dictionary=True)
        cursor.execute("SELECT id, name, age, email FROM students ORDER BY id")
        return cursor.fetchall()
    finally:
        connection.close()

def get_student(student_id):
    connection = get_connection()
    try:
        cursor = connection.cursor(dictionary=True)
        cursor.execute(
            "SELECT id, name, age, email FROM students WHERE id = %s", (student_id,))
        return cursor.fetchone()
    finally:
        connection.close()

def create_student(name, age, email):
    connection = get_connection()
    try:
        cursor = connection.cursor()
        cursor.execute(
            "INSERT INTO students (name, age, email) VALUES (%s, %s, %s)",
            (name, age, email))
        connection.commit()
        new_id = cursor.lastrowid
    except mysql.connector.IntegrityError:
        raise ValueError("a student with email " + email + " already exists")
    finally:
        connection.close()
    return get_student(new_id)

def update_student(student_id, name, age, email):
    connection = get_connection()
    try:
        cursor = connection.cursor()
        cursor.execute(
            "UPDATE students SET name = %s, age = %s, email = %s WHERE id = %s",
            (name, age, email, student_id))
        connection.commit()
    except mysql.connector.IntegrityError:
        raise ValueError("a student with email " + email + " already exists")
    finally:
        connection.close()
    return get_student(student_id)

def delete_student(student_id):
    connection = get_connection()
    try:
        cursor = connection.cursor()
        cursor.execute("DELETE FROM students WHERE id = %s", (student_id,))
        connection.commit()
        return cursor.rowcount > 0
    finally:
        connection.close()

Analyse du code ligne par ligne

Fonction	Ce qu’elle a de nouveau / à remarquer
`get_connection`	Le seul endroit qui ouvre une connexion — et aussi le seul qui connaît les informations de connexion.
`init_db`	`IF NOT EXISTS` → peut être appelée sans risque à chaque démarrage du serveur.
`get_all_students`	`ORDER BY id` pour un ordre stable. Une liste vide est une réponse valide.
`get_student`	`None` quand elle est introuvable. Toutes les autres l’utilisent pour renvoyer « ce qu’il y a dans la base ».
`create_student`	`mysql.connector.IntegrityError` → `ValueError`. `new_id` est enregistré dans le try, et `get_student` est appelée après la fermeture.
`update_student`	Trois états : un dict / `None` / une `ValueError` (l’e-mail est pris). « Introuvable » se décide par `get_student` après la mise à jour — et non par `rowcount`, car dans MySQL une mise à jour avec les mêmes valeurs renvoie un `rowcount` de 0.
`delete_student`	`True` / `False`. Elle ne renvoie pas ce qui a été supprimé (dans l’API : 204, sans corps).

Remarquez la répétition de "SELECT id, name, age, email ..." : dans la pratique en autonomie du thème des bases de données, nous avons vu une constante COLUMNS qui évite la répétition. Ici, nous avons laissé les requêtes en entier, pour que chaque fonction soit lisible seule, pour l’apprentissage. Les deux approches sont légitimes.

Un autre exemple

Un script de test pour les quatre actions — « pratiquer les quatre actions sur une seule entité » du programme. check_crud.py :

import db

connection = db.get_connection()
connection.cursor().execute("DROP TABLE IF EXISTS students")
connection.close()
db.init_db()

print("--- CREATE ---")
dana = db.create_student("Dana", 24, "dana@mail.com")
avi = db.create_student("Avi", 31, "avi@mail.com")
print(dana)
print(avi)
try:
    db.create_student("Dana Copy", 20, "dana@mail.com")
except ValueError as error:
    print("Duplicate rejected:", error)

print("--- READ ---")
print("All:", db.get_all_students())
print("Id 2:", db.get_student(2))
print("Id 99:", db.get_student(99))

print("--- UPDATE ---")
print("Update id 2:", db.update_student(2, "Avi Levi", 32, "avi@mail.com"))
print("Update id 2 again:", db.update_student(2, "Avi Levi", 32, "avi@mail.com"))
print("Update id 99:", db.update_student(99, "Ghost", 30, "ghost@mail.com"))
try:
    db.update_student(2, "Avi Levi", 32, "dana@mail.com")
except ValueError as error:
    print("Email clash rejected:", error)

print("--- DELETE ---")
print("Delete id 1:", db.delete_student(1))
print("Delete id 1 again:", db.delete_student(1))
print("All at the end:", db.get_all_students())

Sortie :

--- CREATE ---
{'id': 1, 'name': 'Dana', 'age': 24, 'email': 'dana@mail.com'}
{'id': 2, 'name': 'Avi', 'age': 31, 'email': 'avi@mail.com'}
Duplicate rejected: a student with email dana@mail.com already exists
--- READ ---
All: [{'id': 1, 'name': 'Dana', 'age': 24, 'email': 'dana@mail.com'}, {'id': 2, 'name': 'Avi', 'age': 31, 'email': 'avi@mail.com'}]
Id 2: {'id': 2, 'name': 'Avi', 'age': 31, 'email': 'avi@mail.com'}
Id 99: None
--- UPDATE ---
Update id 2: {'id': 2, 'name': 'Avi Levi', 'age': 32, 'email': 'avi@mail.com'}
Update id 2 again: {'id': 2, 'name': 'Avi Levi', 'age': 32, 'email': 'avi@mail.com'}
Update id 99: None
Email clash rejected: a student with email dana@mail.com already exists
--- DELETE ---
Delete id 1: True
Delete id 1 again: False
All at the end: [{'id': 2, 'name': 'Avi Levi', 'age': 32, 'email': 'avi@mail.com'}]

Ce qui a été testé : chacune des quatre actions, à la fois dans le cas où tout va bien et dans chaque cas limite : un doublon, un introuvable, une mise à jour répétée avec les mêmes valeurs, un conflit et une suppression répétée. Chaque ligne de la sortie est une ligne du « contrat ». Et ensuite, chacune de ces lignes deviendra une requête HTTP avec un code d’état.

Et ouvrez la table students dans Database Client : il ne reste qu’Avi Levi, avec l’id 2.

Erreur courante

Erreur 1 : un contrat incohérent. get_student renvoie None quand elle ne trouve pas, mais update_student lance une KeyError et delete_student renvoie 0. Le serveur devra gérer chaque fonction différemment. L’état « introuvable » sous la même forme dans tout le module.

Erreur 2 : print dans db.py. print("Student created!") dans create_student. Sur un serveur, cela n’ira nulle part, et dans les tests c’est du « bruit ». La couche de données renvoie des valeurs. C’est celui qui l’appelle qui décide quoi afficher. (Pour un journal → un logger.)

Erreur 3 : supprimer dans le script de test sans réfléchir. DROP TABLE IF EXISTS students en haut de check_crud.py supprime toutes les données. Excellent pour tester, une catastrophe si vous le lancez par erreur sur une vraie base. (Plus tard dans le parcours, vous apprendrez à exécuter les tests sur une base séparée, par exemple school_test.)

Erreur 4 : ne tester que le Create et le Read. Les bugs les plus intéressants se cachent dans UPDATE (l’ordre des paramètres, WHERE) et dans DELETE (rowcount). Les quatre actions, avec les cas limites.

Exercices

Les exercices n’utilisent que ce qui a été vu dans cette leçon et les précédentes. Essayez d’abord seul, et n’ouvrez la solution qu’ensuite.

Exercice 1: Le contrat du module

Écrivez un tableau de « contrat » pour les cinq fonctions de db.py : ce que reçoit chacune, ce qu’elle renvoie, et ce qui se passe quand il n’y a pas de tel étudiant.

get_all_students   get_student   create_student   update_student   delete_student

Afficher la solution

Fonction	Reçoit	Renvoie	S’il n’y a pas de tel étudiant
`get_all_students()`	—	Une liste de dictionnaires	Une liste vide
`get_student(id)`	Un id	Un dictionnaire	`None`
`create_student(name, age, email)`	Les données	Un dictionnaire (avec un id)	— (un e-mail en double → `ValueError`)
`update_student(id, name, age, email)`	Un id + les données	Un dictionnaire mis à jour	`None`
`delete_student(id)`	Un id	`True`	`False`

Un contrat cohérent = le serveur sait exactement à quoi s’attendre de chaque fonction, et comment traduire chaque résultat en code HTTP.

Exercice 2: create_student avec la traduction de l’erreur

Écrivez get_connection, init_db, get_student et create_student, qui traduit un e-mail en double en ValueError.

init_db()
print(create_student("Dana", 24, "dana@mail.com"))
try:
    create_student("Copy", 20, "dana@mail.com")
except ValueError as error:
    print("Duplicate rejected:", error)

Résultat attendu :

{'id': 1, 'name': 'Dana', 'age': 24, 'email': 'dana@mail.com'}
Duplicate rejected: a student with email dana@mail.com already exists

Afficher la solution

import mysql.connector


def get_connection():
    return mysql.connector.connect(
        host="localhost", user="root", password="kodkod", database="school")


def init_db():
    connection = get_connection()
    try:
        cursor = connection.cursor()
        cursor.execute("""
            CREATE TABLE IF NOT EXISTS students (
                id INT AUTO_INCREMENT PRIMARY KEY,
                name VARCHAR(100) NOT NULL,
                age INT NOT NULL,
                email VARCHAR(100) NOT NULL UNIQUE
            )
        """)
    finally:
        connection.close()


def get_student(student_id):
    connection = get_connection()
    try:
        cursor = connection.cursor(dictionary=True)
        cursor.execute("SELECT * FROM students WHERE id = %s", (student_id,))
        return cursor.fetchone()
    finally:
        connection.close()


def create_student(name, age, email):
    connection = get_connection()
    try:
        cursor = connection.cursor()
        cursor.execute(
            "INSERT INTO students (name, age, email) VALUES (%s, %s, %s)",
            (name, age, email))
        connection.commit()
        new_id = cursor.lastrowid
    except mysql.connector.IntegrityError:
        raise ValueError("a student with email " + email + " already exists")
    finally:
        connection.close()
    return get_student(new_id)


init_db()
print(create_student("Dana", 24, "dana@mail.com"))
try:
    create_student("Copy", 20, "dana@mail.com")
except ValueError as error:
    print("Duplicate rejected:", error)

{'id': 1, 'name': 'Dana', 'age': 24, 'email': 'dana@mail.com'}
Duplicate rejected: a student with email dana@mail.com already exists

create_student ne récupère rien elle-même — elle appelle get_student avec le nouvel id. Le finally ferme la connexion même quand une ValueError est lancée.

Exercice 3: Le module complet

Complétez db.py avec get_all_students, update_student et delete_student (selon le contrat), et testez toutes les actions.

init_db()
dana = create_student("Dana", 24, "dana@mail.com")
avi = create_student("Avi", 31, "avi@mail.com")
print(len(get_all_students()))
print(update_student(avi["id"], "Avi Levi", 32, "avi@mail.com"))
print(update_student(99, "X", 1, "x@mail.com"))
print(delete_student(dana["id"]), delete_student(dana["id"]))
print(get_student(dana["id"]))
print(get_all_students())

Résultat attendu :

2
{'id': 2, 'name': 'Avi Levi', 'age': 32, 'email': 'avi@mail.com'}
None
True False
None
[{'id': 2, 'name': 'Avi Levi', 'age': 32, 'email': 'avi@mail.com'}]

Afficher la solution

import mysql.connector


def get_connection():
    return mysql.connector.connect(
        host="localhost", user="root", password="kodkod", database="school")


def init_db():
    connection = get_connection()
    try:
        cursor = connection.cursor()
        cursor.execute("""
            CREATE TABLE IF NOT EXISTS students (
                id INT AUTO_INCREMENT PRIMARY KEY,
                name VARCHAR(100) NOT NULL,
                age INT NOT NULL,
                email VARCHAR(100) NOT NULL UNIQUE
            )
        """)
    finally:
        connection.close()


def get_all_students():
    connection = get_connection()
    try:
        cursor = connection.cursor(dictionary=True)
        cursor.execute("SELECT * FROM students ORDER BY id")
        return cursor.fetchall()
    finally:
        connection.close()


def get_student(student_id):
    connection = get_connection()
    try:
        cursor = connection.cursor(dictionary=True)
        cursor.execute("SELECT * FROM students WHERE id = %s", (student_id,))
        return cursor.fetchone()
    finally:
        connection.close()


def create_student(name, age, email):
    connection = get_connection()
    try:
        cursor = connection.cursor()
        cursor.execute(
            "INSERT INTO students (name, age, email) VALUES (%s, %s, %s)",
            (name, age, email))
        connection.commit()
        new_id = cursor.lastrowid
    except mysql.connector.IntegrityError:
        raise ValueError("a student with email " + email + " already exists")
    finally:
        connection.close()
    return get_student(new_id)


def update_student(student_id, name, age, email):
    connection = get_connection()
    try:
        cursor = connection.cursor()
        cursor.execute(
            "UPDATE students SET name = %s, age = %s, email = %s WHERE id = %s",
            (name, age, email, student_id))
        connection.commit()
    except mysql.connector.IntegrityError:
        raise ValueError("a student with email " + email + " already exists")
    finally:
        connection.close()
    return get_student(student_id)


def delete_student(student_id):
    connection = get_connection()
    try:
        cursor = connection.cursor()
        cursor.execute("DELETE FROM students WHERE id = %s", (student_id,))
        connection.commit()
        return cursor.rowcount > 0
    finally:
        connection.close()


init_db()
dana = create_student("Dana", 24, "dana@mail.com")
avi = create_student("Avi", 31, "avi@mail.com")
print(len(get_all_students()))
print(update_student(avi["id"], "Avi Levi", 32, "avi@mail.com"))
print(update_student(99, "X", 1, "x@mail.com"))
print(delete_student(dana["id"]), delete_student(dana["id"]))
print(get_student(dana["id"]))
print(get_all_students())

2
{'id': 2, 'name': 'Avi Levi', 'age': 32, 'email': 'avi@mail.com'}
None
True False
None
[{'id': 2, 'name': 'Avi Levi', 'age': 32, 'email': 'avi@mail.com'}]

Chaque fonction respecte le contrat : un dictionnaire / None / True / False / une ValueError — sans que rien de mysql.connector ne fuie à l’extérieur.

Exercice 4: Chercher par nom

Pour cet exercice, créez une table simple, avec seulement des noms — Dana, Avi et Daniel :

import mysql.connector

connection = mysql.connector.connect(
    host="localhost", user="root", password="kodkod", database="school")
cursor = connection.cursor()
cursor.execute("DROP TABLE IF EXISTS students")   # start clean, so ids start from 1
cursor.execute("CREATE TABLE students (id INT AUTO_INCREMENT PRIMARY KEY, name VARCHAR(100))")
cursor.execute("INSERT INTO students (name) VALUES ('Dana'), ('Avi'), ('Daniel')")
connection.commit()
connection.close()

Ajoutez search_students(text) qui renvoie une liste de dictionnaires avec tous les étudiants dont le nom contient le texte, avec LIKE et %, triés par id.

print(search_students("an"))

Résultat attendu :

[{'id': 1, 'name': 'Dana'}, {'id': 3, 'name': 'Daniel'}]

Afficher la solution

import mysql.connector


def get_connection():
    return mysql.connector.connect(
        host="localhost", user="root", password="kodkod", database="school")


def search_students(text):
    connection = get_connection()
    try:
        cursor = connection.cursor(dictionary=True)
        cursor.execute(
            "SELECT * FROM students WHERE name LIKE %s ORDER BY id",
            ("%" + text + "%",))
        return cursor.fetchall()
    finally:
        connection.close()


connection = get_connection()
cursor = connection.cursor()
cursor.execute("CREATE TABLE IF NOT EXISTS students "
               "(id INT AUTO_INCREMENT PRIMARY KEY, name VARCHAR(100))")
for name in ["Dana", "Avi", "Daniel"]:
    cursor.execute("INSERT INTO students (name) VALUES (%s)", (name,))
connection.commit()
connection.close()

print(search_students("an"))

[{'id': 1, 'name': 'Dana'}, {'id': 3, 'name': 'Daniel'}]

LIKE compare selon un motif. % = « n’importe quelle suite de caractères ».
Le % est ajouté à la valeur — et non à la commande — pour que la recherche reste aussi à l’abri de l’injection SQL.

Exercice 5: Un script de test

Pourquoi vaut-il la peine d’écrire un script qui teste les cinq fonctions de db.py, et de le lancer sur une table vide ?

Afficher la solution

Un seul test qui couvre tout — après chaque modification du module, vous lancez le script et voyez immédiatement si quelque chose a cassé, au lieu de tester chaque fonction à la main.
Une table vide — les résultats sont prévisibles : le premier id vaut 1, la liste commence vide. Sur une table avec d’anciennes données, le même test renverrait des résultats différents à chaque exécution.
Testez aussi les échecs : un id qui n’existe pas, un e-mail en double, une double suppression — c’est là que se cachent la plupart des bugs.

(Commencez le script par DROP TABLE IF EXISTS students puis init_db() — et de préférence sur une base de test séparée, par exemple school_test, pour ne pas supprimer de vraies données.)

Résumé

db.py = un CRUD complet sur une seule entité : init_db, create, get_all, get, update, delete.
Un contrat cohérent : une liste / un dict / None / True-False / une ValueError. « Introuvable » toujours sous la même forme.
Chaque fonction : sa propre connexion, %s, commit, cursor(dictionary=True), et finally: close().
Pas de HTTP ni de print dans la couche de données. Elle renvoie des valeurs, et celui qui l’appelle décide.
Un script de test qui parcourt les quatre actions et les cas limites. Ensuite, chacune de ses lignes deviendra une requête HTTP.

Docs

kodkod | Cours d'introduction

Title here

Un CRUD complet depuis Python : un module db.py complet et le test des quatre actions

Qu’allons-nous apprendre ?

Pourquoi en avons-nous besoin ?

Explication à partir des bases

Le « contrat » d’une fonction

Les principes communs à toutes les fonctions

Syntaxe

Premier exemple de code

Analyse du code ligne par ligne

Un autre exemple

Erreur courante

Exercices

Exercice 1: Le contrat du module

Exercice 2: create_student avec la traduction de l’erreur

Exercice 3: Le module complet

Exercice 4: Chercher par nom

Exercice 5: Un script de test

Résumé

Documentation officielle

Qu’allons-nous apprendre ?#

Pourquoi en avons-nous besoin ?#

Explication à partir des bases#

Le « contrat » d’une fonction#

Les principes communs à toutes les fonctions#

Syntaxe#

Premier exemple de code#

Analyse du code ligne par ligne#

Un autre exemple#

Erreur courante#

Exercices#

Exercice 1: Le contrat du module#

Exercice 2: create_student avec la traduction de l’erreur#

Exercice 3: Le module complet#

Exercice 4: Chercher par nom#

Exercice 5: Un script de test#

Résumé#

Documentation officielle#

Qu’allons-nous apprendre ?

Pourquoi en avons-nous besoin ?

Explication à partir des bases

Le « contrat » d’une fonction

Les principes communs à toutes les fonctions

Syntaxe

Premier exemple de code

Analyse du code ligne par ligne

Un autre exemple

Erreur courante

Exercices

Exercice 1: Le contrat du module

Exercice 2: create_student avec la traduction de l’erreur

Exercice 3: Le module complet

Exercice 4: Chercher par nom

Exercice 5: Un script de test

Résumé

Documentation officielle