Full CRUD from Python: a Complete db.py Module and Testing All Four Actions

Lesson goals

Assemble a complete db.py module with all the CRUD actions on one Entity
Explain for every function what it returns in every state — a dict, a list, None, a bool or a ValueError
Test all four actions and their edge cases in one script
Be ready to connect the module to a FastAPI server

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`	—
Create	`create_student(name, age, email)`	`INSERT`	A dict (or a `ValueError` on a duplicate email)
Read (all)	`get_all_students()`	`SELECT`	A list of dicts (maybe empty)
Read (one)	`get_student(student_id)`	`SELECT ... WHERE id = %s`	A dict or `None`
Update	`update_student(student_id, name, age, email)`	`UPDATE ... WHERE id = %s`	A dict, `None`, or a `ValueError`
Delete	`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:

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

Principles shared by all the functions

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

Syntax

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

First code example

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

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:

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:

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

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.

get_all_students   get_student   create_student   update_student   delete_student

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

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.

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:

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

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

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.

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:

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

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

Every function keeps the contract: a dictionary / None / True / False / a ValueError — without anything from mysql.connector leaking outside.

Exercise 4: Searching by name

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

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.

print(search_students("an"))

Expected output:

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

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

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?

Show 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.)

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

kodkod | Intro Course

Title here

Full CRUD from Python: a Complete db.py Module and Testing All Four Actions

What will we learn?

Why do we need this?

Explanation from the basics

A function’s “contract”

Principles shared by all the functions

Syntax

First code example

Breaking down the code line by line

Another example

Common mistake

Practice

Exercise 1: The module’s contract

Exercise 2: create_student with translating the error

Exercise 3: The complete module

Exercise 4: Searching by name

Exercise 5: A test script

Summary

Official documentation

What will we learn?#

Why do we need this?#

Explanation from the basics#

A function’s “contract”#

Principles shared by all the functions#

Syntax#

First code example#

Breaking down the code line by line#

Another example#

Common mistake#

Practice#

Exercise 1: The module’s contract#

Exercise 2: create_student with translating the error#

Exercise 3: The complete module#

Exercise 4: Searching by name#

Exercise 5: A test script#

Summary#

Official documentation#

What will we learn?

Why do we need this?

Explanation from the basics

A function’s “contract”

Principles shared by all the functions

Syntax

First code example

Breaking down the code line by line

Another example

Common mistake

Practice

Exercise 1: The module’s contract

Exercise 2: create_student with translating the error

Exercise 3: The complete module

Exercise 4: Searching by name

Exercise 5: A test script

Summary

Official documentation