From 49908634d56934562c32e06c9a54d629aa002d70 Mon Sep 17 00:00:00 2001 From: ptuaqkmj6 Date: Wed, 25 Jun 2025 11:04:29 +0800 Subject: [PATCH] Update sqlite_student_dal.py --- student/dal/sqlite_student_dal.py | 161 ++++++++++++++++++++---------- 1 file changed, 107 insertions(+), 54 deletions(-) diff --git a/student/dal/sqlite_student_dal.py b/student/dal/sqlite_student_dal.py index ba8679c..b1d953a 100644 --- a/student/dal/sqlite_student_dal.py +++ b/student/dal/sqlite_student_dal.py @@ -6,73 +6,93 @@ from datetime import datetime class SQLiteStudentDAL(IStudentDAL): - """学生信息的SQLite数据库存储实现""" + """学生信息的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() + 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, - height INTEGER, - weight REAL, - enrollment_date TEXT, - class_name TEXT, - major TEXT, - email TEXT, - phone TEXT + 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() + conn.commit() # 提交DDL语句 def _row_to_student(self, row: tuple) -> Optional[Student]: - """将数据库行转换为Student对象""" - if not row: + """将数据库查询结果行转换为Student对象 + :param row: 查询结果元组 + :return: 转换后的Student对象或None + 处理逻辑： + - 索引映射：将数据库列索引映射到Student属性 + - 类型转换：日期字符串转date对象，整数转布尔值""" + if not row: # 处理空结果 return None - # 处理日期类型 - enrollment_date = row[7] + # 处理日期类型：从字符串转换为datetime.date + enrollment_date = row[7] # 注意：索引7对应enrollment_date列 if enrollment_date: enrollment_date = datetime.strptime(enrollment_date, '%Y-%m-%d').date() - # 处理布尔类型 - gender = row[3] + # 处理布尔类型：数据库中存储为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] + 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) + 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,)) @@ -80,15 +100,26 @@ class SQLiteStudentDAL(IStudentDAL): 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] + 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() @@ -103,20 +134,26 @@ class SQLiteStudentDAL(IStudentDAL): student.gender, student.height, student.weight, - str(student.enrollment_date) if student.enrollment_date else None, + str(student.enrollment_date) if student.enrollment_date else None, # 日期转字符串 student.class_name, student.major, student.email, student.phone ) ) - conn.commit() + conn.commit() # 提交事务 return True - except sqlite3.IntegrityError: # 违反唯一约束 + 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( @@ -129,27 +166,34 @@ class SQLiteStudentDAL(IStudentDAL): student.gender, student.height, student.weight, - str(student.enrollment_date) if student.enrollment_date else None, + str(student.enrollment_date) if student.enrollment_date else None, # 日期转字符串 student.class_name, student.major, student.email, student.phone, - student.id_card + student.id_card # 条件：根据身份证号更新 ) ) conn.commit() - return cursor.rowcount > 0 + 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 + 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,)) @@ -157,7 +201,12 @@ class SQLiteStudentDAL(IStudentDAL): 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 + '%',)) @@ -165,7 +214,9 @@ class SQLiteStudentDAL(IStudentDAL): 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 + '%',)) @@ -173,9 +224,11 @@ class SQLiteStudentDAL(IStudentDAL): 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] \ No newline at end of file + return [self._row_to_student(row) for row in rows]