import sqlite3 from typing import List, Optional from .student_dal import IStudentDAL from ..model.student import Student from datetime import datetime class SQLiteStudentDAL(IStudentDAL): """学生信息的SQLite数据库存储实现，实现了IStudentDAL接口，提供基于SQLite的学生数据持久化功能""" def __init__(self, db_path: str = 'students.db'): """初始化SQLite数据访问层 :param db_path: 数据库文件路径，默认使用当前目录下的students.db""" self.db_path = db_path self._create_table() # 确保学生表存在，不存在则创建 def _create_table(self): """创建学生信息表（如果不存在） 表结构设计： - id：自增主键 - id_card/stu_id：设置唯一约束，确保数据唯一性 - enrollment_date：存储为YYYY-MM-DD格式字符串 - gender：布尔值存储为整数（0/1）""" with sqlite3.connect(self.db_path) as conn: cursor = conn.cursor() cursor.execute(''' CREATE TABLE IF NOT EXISTS students ( id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT NOT NULL, -- 学生姓名，非空 id_card TEXT UNIQUE NOT NULL, -- 身份证号，唯一且非空 stu_id TEXT UNIQUE NOT NULL, -- 学号，唯一且非空 gender INTEGER, -- 性别（0/1） height INTEGER, -- 身高（厘米） weight REAL, -- 体重（公斤） enrollment_date TEXT, -- 入学日期（YYYY-MM-DD） class_name TEXT, -- 班级名称 major TEXT, -- 专业 email TEXT, -- 邮箱 phone TEXT -- 电话 ) ''') conn.commit() # 提交DDL语句 def _row_to_student(self, row: tuple) -> Optional[Student]: """将数据库查询结果行转换为Student对象 :param row: 查询结果元组 :return: 转换后的Student对象或None 处理逻辑： - 索引映射：将数据库列索引映射到Student属性 - 类型转换：日期字符串转date对象，整数转布尔值""" if not row: # 处理空结果 return None # 处理日期类型：从字符串转换为datetime.date enrollment_date = row[7] # 注意：索引7对应enrollment_date列 if enrollment_date: enrollment_date = datetime.strptime(enrollment_date, '%Y-%m-%d').date() # 处理布尔类型：数据库中存储为0/1，转换为True/False gender = row[4] # 注意：索引4对应gender列 if gender is not None: gender = bool(gender) # 构建并返回Student对象 return Student( name=row[1], # 姓名 id_card=row[2], # 身份证号 stu_id=row[3], # 学号 gender=gender, # 性别（布尔值） height=row[4], # 身高 weight=row[5], # 体重 enrollment_date=enrollment_date, # 入学日期 class_name=row[8], # 班级 major=row[9], # 专业 email=row[10], # 邮箱 phone=row[11] # 电话 ) def get_by_id(self, id_card: str) -> Optional[Student]: """根据身份证号查询学生信息 :param id_card: 待查询的身份证号 :return: 匹配的Student对象或None 实现逻辑： - 使用参数化查询防止SQL注入 - 通过id_card唯一索引快速定位记录""" with sqlite3.connect(self.db_path) as conn: cursor = conn.cursor() cursor.execute("SELECT * FROM students WHERE id_card = ?", (id_card,)) row = cursor.fetchone() # 获取单条记录 return self._row_to_student(row) # 转换为Student对象 def get_by_stu_id(self, stu_id: str) -> Optional[Student]: """根据学号查询学生信息，逻辑同get_by_id :param stu_id: 待查询的学号 :return: 匹配的Student对象或None""" with sqlite3.connect(self.db_path) as conn: cursor = conn.cursor() cursor.execute("SELECT * FROM students WHERE stu_id = ?", (stu_id,)) row = cursor.fetchone() return self._row_to_student(row) def get_all(self) -> List[Student]: """获取所有学生信息 :return: 包含所有学生的列表 实现逻辑： - 查询全量数据 - 批量转换为Student对象""" with sqlite3.connect(self.db_path) as conn: cursor = conn.cursor() cursor.execute("SELECT * FROM students") rows = cursor.fetchall() # 获取所有记录 return [self._row_to_student(row) for row in rows] # 列表推导式批量转换 def add(self, student: Student) -> bool: """添加学生信息 :param student: 待添加的Student对象 :return: 添加成功返回True，失败返回False 业务逻辑： 1. 使用事务确保数据一致性 2. 自动处理日期类型转换 3. 利用数据库唯一约束防止重复数据 4. 异常处理：IntegrityError表示违反唯一性约束""" try: with sqlite3.connect(self.db_path) as conn: cursor = conn.cursor() cursor.execute( """INSERT INTO students (name, id_card, stu_id, gender, height, weight, enrollment_date, class_name, major, email, phone) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""", ( student.name, student.id_card, student.stu_id, student.gender, student.height, student.weight, str(student.enrollment_date) if student.enrollment_date else None, # 日期转字符串 student.class_name, student.major, student.email, student.phone ) ) conn.commit() # 提交事务 return True except sqlite3.IntegrityError: # 违反唯一约束（身份证号或学号重复） return False def update(self, student: Student) -> bool: """更新学生信息 :param student: 包含更新信息的Student对象 :return: 更新成功返回True，失败返回False 实现逻辑： 1. 以身份证号作为唯一标识 2. 返回受影响的行数判断操作是否成功 3. 自动处理日期类型转换""" with sqlite3.connect(self.db_path) as conn: cursor = conn.cursor() cursor.execute( """UPDATE students SET name=?, gender=?, height=?, weight=?, enrollment_date=?, class_name=?, major=?, email=?, phone=? WHERE id_card=?""", ( student.name, student.gender, student.height, student.weight, str(student.enrollment_date) if student.enrollment_date else None, # 日期转字符串 student.class_name, student.major, student.email, student.phone, student.id_card # 条件：根据身份证号更新 ) ) conn.commit() return cursor.rowcount > 0 # 判断是否有记录被更新 def delete_by_id(self, id_card: str) -> bool: """根据身份证号删除学生信息 :param id_card: 待删除学生的身份证号 :return: 删除成功返回True，失败返回False 实现逻辑： - 通过受影响行数判断删除是否成功 - 使用事务确保操作原子性""" with sqlite3.connect(self.db_path) as conn: cursor = conn.cursor() cursor.execute("DELETE FROM students WHERE id_card = ?", (id_card,)) conn.commit() return cursor.rowcount > 0 # 判断是否有记录被删除 def delete_by_stu_id(self, stu_id: str) -> bool: """根据学号删除学生信息，逻辑同delete_by_id :param stu_id: 待删除学生的学号 :return: 删除成功返回True，失败返回False""" with sqlite3.connect(self.db_path) as conn: cursor = conn.cursor() cursor.execute("DELETE FROM students WHERE stu_id = ?", (stu_id,)) conn.commit() return cursor.rowcount > 0 def search_by_name(self, name: str) -> List[Student]: """根据姓名模糊查询学生信息 :param name: 待查询的姓名片段 :return: 包含匹配学生的列表 实现逻辑： - 使用LIKE '%name%'实现模糊匹配 - 支持空字符串查询（返回所有记录）""" with sqlite3.connect(self.db_path) as conn: cursor = conn.cursor() cursor.execute("SELECT * FROM students WHERE name LIKE ?", ('%' + name + '%',)) rows = cursor.fetchall() return [self._row_to_student(row) for row in rows] def search_by_class(self, class_name: str) -> List[Student]: """根据班级模糊查询学生信息，逻辑同search_by_name :param class_name: 待查询的班级片段 :return: 包含匹配学生的列表""" with sqlite3.connect(self.db_path) as conn: cursor = conn.cursor() cursor.execute("SELECT * FROM students WHERE class_name LIKE ?", ('%' + class_name + '%',)) rows = cursor.fetchall() return [self._row_to_student(row) for row in rows] def search_by_major(self, major: str) -> List[Student]: """根据专业模糊查询学生信息，逻辑同search_by_name :param major: 待查询的专业片段 :return: 包含匹配学生的列表""" with sqlite3.connect(self.db_path) as conn: cursor = conn.cursor() cursor.execute("SELECT * FROM students WHERE major LIKE ?", ('%' + major + '%',)) rows = cursor.fetchall() return [self._row_to_student(row) for row in rows]