# Un CRUD complet depuis Python : un module db.py complet et le test des quatre actions<no value>

## 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` | — |
| **C**reate | `create_student(name, age, email)` | `INSERT` | Un dict (ou une `ValueError` sur un e-mail en double) |
| **R**ead (tous) | `get_all_students()` | `SELECT` | Une liste de dicts (peut-être vide) |
| **R**ead (un) | `get_student(student_id)` | `SELECT ... WHERE id = %s` | Un dict ou `None` |
| **U**pdate | `update_student(student_id, name, age, email)` | `UPDATE ... WHERE id = %s` | Un dict, `None`, ou une `ValueError` |
| **D**elete | `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 :

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

### Les principes communs à toutes les fonctions

1. **Une connexion pour chaque action** — `get_connection()`, et `close()` dans `finally`.
2. **Des paramètres uniquement** — `%s`, jamais de concaténation.
3. **commit** après chaque modification.
4. **Un dict en sortie** — `cursor(dictionary=True)`.
5. **« Introuvable » = `None` / `False`**, et non une exception.
6. **Une violation de contrainte = une `ValueError`** avec un message clair.
7. **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 :

```text
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`) :

```python
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` :

```python
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 :**

```text
--- 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.**

<!-- exercises:start — generated by scripts/intro6/sync_site_exercises.py; edit scripts/intro6/exercises/ -->
## 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.

```text
get_all_students   get_student   create_student   update_student   delete_student
```

{{< 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.
{{< /solution >}}

### 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`.

```python
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 :**

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

{{< solution >}}
```python
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)
```

```text
{'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.
{{< /solution >}}

### 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.

```python
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 :**

```text
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'}]
```

{{< solution >}}
```python
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())
```

```text
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.
{{< /solution >}}

### Exercice 4: Chercher par nom

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

```py
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`.

```python
print(search_students("an"))
```

**Résultat attendu :**

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

{{< solution >}}
```python
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"))
```

```text
[{'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.
{{< /solution >}}

### 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** ?

{{< 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.)
{{< /solution >}}

<!-- exercises:end -->

## 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:start — generated by scripts/intro6/site_docs.py; edit the links there -->

## Documentation officielle

- [Connector/Python — Coding Examples](https://dev.mysql.com/doc/connector-python/en/connector-python-examples.html)
- [Python Tutorial — Modules](https://docs.python.org/3/tutorial/modules.html)

<!-- docs:end -->