ptuaqkmj6
/
abc
1
0
Fork 0
Code Issues Pull Requests 1 Packages Projects Releases Wiki Activity

Update sqlite_student_dal.py

Browse Source
master
ptuaqkmj6 2 months ago
parent 6ddcb003d5
commit 49908634d5
1 changed files with 107 additions and 54 deletions
Show Stats Download Patch File Download Diff File

161
student/dal/sqlite_student_dal.py
Unescape View File

@ -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]
return [self._row_to_student(row) for row in rows]

Write Preview
Loading…
Cancel
Save