# Full CRUD from Python: a Complete db.py Module and Testing All Four Actions<no value>

## What will we learn?

At the end of this lesson, as the syllabus defines, we have **full CRUD from Python**. The lesson assembles all the parts from the previous lessons into **one module**, `db.py`, and tests it from end to end.

| CRUD | Function | SQL | Returns |
|---|---|---|---|
| (preparation) | `init_db()` | `CREATE TABLE IF NOT EXISTS` | — |
| **C**reate | `create_student(name, age, email)` | `INSERT` | A dict (or a `ValueError` on a duplicate email) |
| **R**ead (all) | `get_all_students()` | `SELECT` | A list of dicts (maybe empty) |
| **R**ead (one) | `get_student(student_id)` | `SELECT ... WHERE id = %s` | A dict or `None` |
| **U**pdate | `update_student(student_id, name, age, email)` | `UPDATE ... WHERE id = %s` | A dict, `None`, or a `ValueError` |
| **D**elete | `delete_student(student_id)` | `DELETE ... WHERE id = %s` | `True` / `False` |

## Why do we need this?

Next we connect this module to a server. Every Endpoint will be **a few lines**, which call one function from here and translate the result into HTTP. For this to work without surprises, the module needs to be **complete, consistent and tested**, with a clear "contract": what every function returns in every state.

## Explanation from the basics

### A function's "contract"

**Contract** = the function's promise to whoever calls it: what it receives, what it returns, and which errors it may throw. The table above **is** the contract of `db.py`. The server will rely on it:

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

### Principles shared by all the functions

1. **A connection for every action** — `get_connection()`, and `close()` in `finally`.
2. **Parameters only** — `%s`, never concatenation.
3. **commit** after every change.
4. **A dict going out** — `cursor(dictionary=True)`.
5. **"Not found" = `None` / `False`**, and not an Exception.
6. **A constraint violation = a `ValueError`** with a clear message.
7. **No HTTP, no `print`** — the module does not know who uses it (a server, a script, a menu).

## Syntax

The structure of the 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
```

## First code example

**The complete `db.py`** (in the project `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()
```

## Breaking down the code line by line

| Function | What is new in it / what to notice |
|---|---|
| `get_connection` | The **only** place that opens a connection — and also the only one that knows the connection details. |
| `init_db` | `IF NOT EXISTS` → safe to call on every start of the server. |
| `get_all_students` | `ORDER BY id` for a stable order. An empty list is a valid answer. |
| `get_student` | `None` when not found. All the others use it to return "what is in the database". |
| `create_student` | `mysql.connector.IntegrityError` → `ValueError`. `new_id` is saved **inside** the try, and `get_student` is called **after** closing. |
| `update_student` | Three states: a dict / `None` / a `ValueError` (the email is taken). "Not found" is decided by `get_student` after the update — **not** by `rowcount`, because in MySQL an update to the same values returns a `rowcount` of 0. |
| `delete_student` | `True` / `False`. It does not return what was deleted (in the API: 204, without a body). |

**Notice the repetition of `"SELECT id, name, age, email ..."`:** in the independent practice of the databases topic we saw a constant `COLUMNS` that prevents the repetition. Here we left the queries in full, so that every function is **readable on its own**, for learning. Both approaches are legitimate.

## Another example

**A test script for all four actions** — "practicing all four actions on one Entity" from the syllabus. `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())
```

**Output:**

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

**What was tested:** each of the four actions, **both** on the happy path **and** in every edge case: a duplicate, not found, a repeated update to the same values, a clash and a repeated delete. Every line in the output is a line in the "contract". **And next, every such line will become an HTTP request with a status code.**

**And open the `students` table in Database Client:** only Avi Levi is left, with id 2.

## Common mistake

**Mistake 1: an inconsistent contract.** `get_student` returns `None` when not found, but `update_student` throws a `KeyError` and `delete_student` returns `0`. The server will have to handle every function differently. **The "not found" state in the same form throughout the module.**

**Mistake 2: `print` inside `db.py`.** `print("Student created!")` inside `create_student`. On a server this will go nowhere, and in tests it is "noise". The data layer **returns** values. Whoever calls it decides what to show. (For a log → a `logger`.)

**Mistake 3: deleting in the test script without thinking.** `DROP TABLE IF EXISTS students` at the top of `check_crud.py` deletes **all the data**. Excellent for testing, **a disaster** if you run it by mistake against a real database. (Later in the track you will learn to run tests on a separate database, for example `school_test`.)

**Mistake 4: testing only the Create and the Read.** The most interesting bugs hide in UPDATE (the order of parameters, WHERE) and in DELETE (rowcount). **All four actions, with edge cases.**

<!-- exercises:start — generated by scripts/intro6/sync_site_exercises.py; edit scripts/intro6/exercises/ -->
## Practice

The exercises use only what was taught in this lesson and the lessons before it. Try on your own first, and only then open the solution.

### Exercise 1: The module's contract

Write a "contract" table for the five functions of `db.py`: what each one receives, what it returns, and what happens when there is no such student.

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

{{< solution >}}
| Function | Receives | Returns | If there is no such student |
|---|---|---|---|
| `get_all_students()` | — | A list of dictionaries | An empty list |
| `get_student(id)` | An id | A dictionary | `None` |
| `create_student(name, age, email)` | The data | A dictionary (with an id) | — (a duplicate email → `ValueError`) |
| `update_student(id, name, age, email)` | An id + the data | An updated dictionary | `None` |
| `delete_student(id)` | An id | `True` | `False` |

A consistent contract = the server knows exactly what to expect from every function, and how to translate every result into an HTTP code.
{{< /solution >}}

### Exercise 2: create_student with translating the error

Write `get_connection`, `init_db`, `get_student` and `create_student`, which translates a duplicate email into a `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)
```

**Expected output:**

```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` does not fetch by itself — it calls `get_student` with the new id. The `finally` closes the connection even when a `ValueError` is thrown.
{{< /solution >}}

### Exercise 3: The complete module

Complete `db.py` with `get_all_students`, `update_student` and `delete_student` (according to the contract), and test all the 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())
```

**Expected output:**

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

Every function keeps the contract: a dictionary / `None` / `True` / `False` / a `ValueError` — without anything from `mysql.connector` leaking outside.
{{< /solution >}}

### Exercise 4: Searching by name

For this exercise create a simple table, with names only — Dana, Avi and 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()
```

Add `search_students(text)` that returns **a list of dictionaries** with all the students whose name **contains** the text, using `LIKE` and `%`, sorted by `id`.

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

**Expected output:**

```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` compares by a pattern. `%` = "any sequence of characters".
- The `%` is added to the **value** — not to the command — so that the search also stays safe from SQL Injection.
{{< /solution >}}

### Exercise 5: A test script

Why is it worth writing a script that tests all five functions of `db.py`, and running it on **an empty table**?

{{< solution >}}
- **One test that covers everything** — after every change in the module you run the script and immediately see whether something broke, instead of testing every function by hand.
- **An empty table** — the results are **predictable**: the first id is 1, the list starts empty. On a table with old data, the same test would return different results on every run.
- **Test failures too**: an id that does not exist, a duplicate email, a double delete — that is where most bugs hide.

(Start the script with `DROP TABLE IF EXISTS students` and then `init_db()` — and preferably on a separate test database, for example `school_test`, so as not to delete real data.)
{{< /solution >}}

<!-- exercises:end -->

## Summary

- **`db.py` = full CRUD on one Entity**: `init_db`, `create`, `get_all`, `get`, `update`, `delete`.
- **A consistent contract**: a list / a dict / `None` / `True`-`False` / a `ValueError`. "Not found" always in the same form.
- Every function: its own connection, `%s`, commit, `cursor(dictionary=True)`, and `finally: close()`.
- **No HTTP and no print** in the data layer. It returns values, and whoever calls it decides.
- A test script that goes over **all four actions and the edge cases**. Next, every line in it will become an HTTP request.

<!-- docs:start — generated by scripts/intro6/site_docs.py; edit the links there -->

## Official documentation

- [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 -->