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