AI0814_jiaoan_public/w0_d6/07_students_mysql_pymysql.py

"""
07 连接数据库版本：学生增删改查

数据库连接信息：
    host: localhost
    port: 3306
    user: root
    password: 123456
    database: student_demo

依赖安装：
    pip install pymysql

本文件会自动创建数据库和 students 表。
"""

import pymysql


DB_CONFIG = {
    "host": "localhost",
    "port": 3306,
    "user": "root",
    "password": "123456",
    "charset": "utf8mb4",
}

DATABASE = "student_demo"


def get_server_connection():
    return pymysql.connect(**DB_CONFIG)


def get_db_connection():
    return pymysql.connect(
        **DB_CONFIG,
        database=DATABASE,
        cursorclass=pymysql.cursors.DictCursor,
    )


def init_database():
    with get_server_connection() as connection:
        with connection.cursor() as cursor:
            cursor.execute(
                f"CREATE DATABASE IF NOT EXISTS {DATABASE} "
                "DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci"
            )
        connection.commit()

    with get_db_connection() as connection:
        with connection.cursor() as cursor:
            cursor.execute(
                """
                CREATE TABLE IF NOT EXISTS students (
                    id INT PRIMARY KEY AUTO_INCREMENT,
                    name VARCHAR(50) NOT NULL,
                    age INT NOT NULL,
                    phone VARCHAR(30) NOT NULL
                )
                """
            )
        connection.commit()


class StudentRepository:
    def add(self, name, age, phone):
        sql = "INSERT INTO students(name, age, phone) VALUES(%s, %s, %s)"
        with get_db_connection() as connection:
            with connection.cursor() as cursor:
                cursor.execute(sql, (name, age, phone))
            connection.commit()

    def list_all(self):
        with get_db_connection() as connection:
            with connection.cursor() as cursor:
                cursor.execute("SELECT id, name, age, phone FROM students ORDER BY id")
                return cursor.fetchall()

    def search_by_name(self, name):
        with get_db_connection() as connection:
            with connection.cursor() as cursor:
                cursor.execute(
                    "SELECT id, name, age, phone FROM students WHERE name = %s ORDER BY id",
                    (name,),
                )
                return cursor.fetchall()

    def update(self, student_id, name, age, phone):
        sql = "UPDATE students SET name = %s, age = %s, phone = %s WHERE id = %s"
        with get_db_connection() as connection:
            with connection.cursor() as cursor:
                affected_rows = cursor.execute(sql, (name, age, phone, student_id))
            connection.commit()
        return affected_rows

    def delete(self, student_id):
        with get_db_connection() as connection:
            with connection.cursor() as cursor:
                affected_rows = cursor.execute(
                    "DELETE FROM students WHERE id = %s",
                    (student_id,),
                )
            connection.commit()
        return affected_rows


class StudentApp:
    def __init__(self):
        self.repository = StudentRepository()

    def show_menu(self):
        print("\n====== 学生管理系统：PyMySQL 数据库版本 ======")
        print("1. 添加学生")
        print("2. 查看所有学生")
        print("3. 查询学生")
        print("4. 修改学生")
        print("5. 删除学生")
        print("0. 退出")

    def add_student(self):
        name = input("姓名：").strip()
        age = int(input("年龄：").strip())
        phone = input("电话：").strip()
        self.repository.add(name, age, phone)
        print("添加成功")

    def list_students(self):
        students = self.repository.list_all()
        if not students:
            print("暂无学生")
            return
        for student in students:
            print(student)

    def search_student(self):
        name = input("请输入要查询的姓名：").strip()
        students = self.repository.search_by_name(name)
        if not students:
            print("没有找到该学生")
            return
        for student in students:
            print(student)

    def update_student(self):
        student_id = int(input("请输入要修改的学生 ID：").strip())
        name = input("新姓名：").strip()
        age = int(input("新年龄：").strip())
        phone = input("新电话：").strip()
        affected_rows = self.repository.update(student_id, name, age, phone)

        if affected_rows:
            print("修改成功")
        else:
            print("没有找到该学生")

    def delete_student(self):
        student_id = int(input("请输入要删除的学生 ID：").strip())
        affected_rows = self.repository.delete(student_id)

        if affected_rows:
            print("删除成功")
        else:
            print("没有找到该学生")

    def run(self):
        menu = {
            "1": self.add_student,
            "2": self.list_students,
            "3": self.search_student,
            "4": self.update_student,
            "5": self.delete_student,
        }

        while True:
            self.show_menu()
            choice = input("请输入操作编号：").strip()

            if choice == "0":
                print("程序结束")
                break

            action = menu.get(choice)
            if action is None:
                print("输入有误，请重新输入")
                continue

            try:
                action()
            except ValueError:
                print("年龄和 ID 必须输入数字")


if __name__ == "__main__":
    init_database()
    app = StudentApp()
    app.run()