Полный CRUD из Python: законченный модуль db.py и проверка всех четырёх действий

Цели урока

Собрать законченный модуль db.py со всеми действиями CRUD над одной сущностью (Entity)
Объяснять для каждой функции, что она возвращает в каждом состоянии — dict, список, None, bool или ValueError
Проверять все четыре действия и их граничные случаи одним скриптом
Быть готовыми подключить модуль к серверу FastAPI

Что мы изучим?

В конце этого урока, как определяет программа курса, у нас есть полный CRUD из Python. Урок собирает все части из предыдущих уроков в один модуль db.py и проверяет его от начала до конца.

CRUD	Функция	SQL	Возвращает
(подготовка)	`init_db()`	`CREATE TABLE IF NOT EXISTS`	—
Create	`create_student(name, age, email)`	`INSERT`	dict (или `ValueError` при дублирующемся email)
Read (все)	`get_all_students()`	`SELECT`	Список словарей (может быть пустым)
Read (один)	`get_student(student_id)`	`SELECT ... WHERE id = %s`	dict или `None`
Update	`update_student(student_id, name, age, email)`	`UPDATE ... WHERE id = %s`	dict, `None` или `ValueError`
Delete	`delete_student(student_id)`	`DELETE ... WHERE id = %s`	`True` / `False`

Зачем это нужно?

Дальше мы подключим этот модуль к серверу. Каждый Endpoint будет несколькими строками, которые вызывают одну функцию отсюда и переводят результат в HTTP. Чтобы это работало без сюрпризов, модуль должен быть законченным, согласованным и проверенным, с понятным «контрактом»: что каждая функция возвращает в каждом состоянии.

Объяснение с основ

«Контракт» функции

Контракт (Contract) = обещание функции тому, кто её вызывает: что она получает, что возвращает и какие ошибки может выбросить. Таблица выше и есть контракт db.py. Сервер будет на него опираться:

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

Общие принципы всех функций

Соединение на каждое действие — get_connection() и close() в finally.
Только параметры — %s, никогда не конкатенация.
commit после каждого изменения.
На выходе dict — cursor(dictionary=True).
«Не найдено» = None / False, а не исключение.
Нарушение ограничения = ValueError с понятным сообщением.
Никакого HTTP, никакого print — модуль не знает, кто его использует (сервер, скрипт, меню).

Синтаксис

Структура модуля:

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

Первый пример кода

Законченный db.py (в проекте 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()

Разбор кода построчно

Функция	Что в ней нового / на что обратить внимание
`get_connection`	Единственное место, которое открывает соединение, — и единственное, которое знает данные для подключения.
`init_db`	`IF NOT EXISTS` → безопасно вызывать при каждом запуске сервера.
`get_all_students`	`ORDER BY id` для стабильного порядка. Пустой список — допустимый ответ.
`get_student`	`None`, когда не найдено. Все остальные используют её, чтобы вернуть «то, что есть в базе».
`create_student`	`mysql.connector.IntegrityError` → `ValueError`. `new_id` сохраняется внутри try, а `get_student` вызывается после закрытия.
`update_student`	Три состояния: dict / `None` / `ValueError` (email занят). «Не найдено» решает `get_student` после обновления — а не `rowcount`, потому что в MySQL обновление на те же значения возвращает `rowcount` 0.
`delete_student`	`True` / `False`. Не возвращает удалённое (в API: 204, без тела).

Обратите внимание на повторение "SELECT id, name, age, email ...": в самостоятельной практике темы баз данных мы видели константу COLUMNS, которая предотвращает повторение. Здесь мы оставили запросы полностью, чтобы каждая функция была читаема сама по себе, для обучения. Оба подхода законны.

Ещё один пример

Проверочный скрипт для всех четырёх действий — «практика всех четырёх действий над одной сущностью» из программы курса. 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())

Вывод:

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

Что проверено: каждое из четырёх действий — и на «счастливом пути», и в каждом граничном случае: дубликат, не найдено, повторное обновление на те же значения, конфликт и повторное удаление. Каждая строка вывода — строка «контракта». А дальше каждая такая строка станет HTTP-запросом с кодом состояния.

И откройте таблицу students в Database Client: остался только Avi Levi с id 2.

Типичная ошибка

Ошибка 1: несогласованный контракт. get_student возвращает None, когда не найдено, а update_student выбрасывает KeyError, и delete_student возвращает 0. Серверу придётся обрабатывать каждую функцию по-своему. Состояние «не найдено» — в одном и том же виде во всём модуле.

Ошибка 2: print внутри db.py. print("Student created!") внутри create_student. На сервере это никуда не попадёт, а в проверках это «шум». Слой данных возвращает значения. Тот, кто его вызывает, решает, что показать. (Для лога → logger.)

Ошибка 3: удаление в проверочном скрипте без размышлений. DROP TABLE IF EXISTS students в начале check_crud.py удаляет все данные. Отлично для проверки, катастрофа, если запустить по ошибке на настоящей базе. (Позже в треке вы научитесь запускать проверки на отдельной базе, например school_test.)

Ошибка 4: проверять только Create и Read. Самые интересные баги прячутся в UPDATE (порядок параметров, WHERE) и в DELETE (rowcount). Все четыре действия, с граничными случаями.

Практика

Упражнения опираются только на то, что пройдено в этом уроке и в предыдущих. Сначала попробуйте сами, и только потом открывайте решение.

Упражнение 1: Контракт модуля

Составьте таблицу «контракта» для пяти функций db.py: что каждая получает, что возвращает и что происходит, когда такого студента нет.

get_all_students   get_student   create_student   update_student   delete_student

Показать решение

Функция	Получает	Возвращает	Если такого студента нет
`get_all_students()`	—	Список словарей	Пустой список
`get_student(id)`	id	Словарь	`None`
`create_student(name, age, email)`	Данные	Словарь (с id)	— (дублирующийся email → `ValueError`)
`update_student(id, name, age, email)`	id + данные	Обновлённый словарь	`None`
`delete_student(id)`	id	`True`	`False`

Согласованный контракт = сервер точно знает, чего ожидать от каждой функции и как перевести каждый результат в код HTTP.

Упражнение 2: create_student с переводом ошибки

Напишите get_connection, init_db, get_student и create_student, которая переводит дублирующийся email в 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)

Ожидаемый вывод:

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

Показать решение

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 не делает выборку сама — она вызывает get_student с новым id. finally закрывает соединение, даже когда выброшен ValueError.

Упражнение 3: Законченный модуль

Дополните db.py функциями get_all_students, update_student и delete_student (по контракту) и проверьте все действия.

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

Показать решение

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

Каждая функция соблюдает контракт: словарь / None / True / False / ValueError — и ничего из mysql.connector не просачивается наружу.

Упражнение 4: Поиск по имени

Для этого упражнения создайте простую таблицу только с именами — Дана, Ави и Даниэль:

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()

Добавьте search_students(text), которая возвращает список словарей со всеми студентами, чьё имя содержит текст, с помощью LIKE и %, отсортированный по id.

print(search_students("an"))

Ожидаемый вывод:

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

Показать решение

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 сравнивает по шаблону. % = «любая последовательность символов».
% добавляется к значению, а не к команде, — так поиск тоже остаётся защищённым от SQL Injection.

Упражнение 5: Проверочный скрипт

Почему стоит написать скрипт, который проверяет все пять функций db.py, и запускать его на пустой таблице?

Показать решение

Одна проверка, которая покрывает всё — после каждого изменения в модуле вы запускаете скрипт и сразу видите, не сломалось ли что-то, вместо того чтобы проверять каждую функцию вручную.
Пустая таблица — результаты предсказуемы: первый id — 1, список начинается пустым. На таблице со старыми данными та же проверка возвращала бы разные результаты при каждом запуске.
Проверяйте и неудачи: несуществующий id, дублирующийся email, двойное удаление — именно там прячется большинство багов.

(Начните скрипт с DROP TABLE IF EXISTS students, а затем init_db() — и лучше на отдельной проверочной базе, например school_test, чтобы не удалить настоящие данные.)

Итоги

db.py = полный CRUD над одной сущностью: init_db, create, get_all, get, update, delete.
Согласованный контракт: список / dict / None / True-False / ValueError. «Не найдено» — всегда в одном и том же виде.
Каждая функция: своё соединение, %s, commit, cursor(dictionary=True) и finally: close().
Никакого HTTP и никакого print в слое данных. Он возвращает значения, а решает тот, кто его вызывает.
Проверочный скрипт, который проходит все четыре действия и граничные случаи. Дальше каждая его строка станет HTTP-запросом.

Docs

kodkod | Вводный курс

Title here

Полный CRUD из Python: законченный модуль db.py и проверка всех четырёх действий

Что мы изучим?

Зачем это нужно?

Объяснение с основ

«Контракт» функции

Общие принципы всех функций

Синтаксис

Первый пример кода

Разбор кода построчно

Ещё один пример

Типичная ошибка

Практика

Упражнение 1: Контракт модуля

Упражнение 2: create_student с переводом ошибки

Упражнение 3: Законченный модуль

Упражнение 4: Поиск по имени

Упражнение 5: Проверочный скрипт

Итоги

Официальная документация

Что мы изучим?#

Зачем это нужно?#

Объяснение с основ#

«Контракт» функции#

Общие принципы всех функций#

Синтаксис#

Первый пример кода#

Разбор кода построчно#

Ещё один пример#

Типичная ошибка#

Практика#

Упражнение 1: Контракт модуля#

Упражнение 2: create_student с переводом ошибки#

Упражнение 3: Законченный модуль#

Упражнение 4: Поиск по имени#

Упражнение 5: Проверочный скрипт#

Итоги#

Официальная документация#

Что мы изучим?

Зачем это нужно?

Объяснение с основ

«Контракт» функции

Общие принципы всех функций

Синтаксис

Первый пример кода

Разбор кода построчно

Ещё один пример

Типичная ошибка

Практика

Упражнение 1: Контракт модуля

Упражнение 2: create_student с переводом ошибки

Упражнение 3: Законченный модуль

Упражнение 4: Поиск по имени

Упражнение 5: Проверочный скрипт

Итоги

Официальная документация