CRUD מלא מתוך Python: מודול db.py שלם ובדיקת כל ארבע הפעולות

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

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

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

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

get_all_students   get_student   create_student   update_student   delete_student

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

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

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

print(search_students("an"))

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