153 lines
5.8 KiB
Python
153 lines
5.8 KiB
Python
# 导入 SQLAlchemy 的 Session 类，用于类型提示
|
||
from sqlalchemy.orm import Session
|
||
# 导入学生信息模型和性别枚举
|
||
from model.student_info_region_model import Student_info, GenderEnum
|
||
# 导入完整性错误异常，用于捕获外键/唯一约束等数据库错误
|
||
from sqlalchemy.exc import IntegrityError
|
||
|
||
|
||
# ==================== 条件查询学生信息 ====================
|
||
def get_part_student_info(session: Session, student_id: str, student_name: str, gender: str):
|
||
"""
|
||
根据学号、姓名、性别查询学生信息（均为可选条件）。
|
||
:param session: 数据库会话
|
||
:param student_id: 学号，精确匹配
|
||
:param student_name: 姓名，精确匹配
|
||
:param gender: 性别，前端传“男/女”，内部转为“1/0”
|
||
:return: 学生对象列表
|
||
"""
|
||
# 前端传来的性别为 男/女，与数据库的枚举值（'0'/'1'）不符，需要转换
|
||
# 注意：若 gender 为 None（不筛选），此处会被错误地转为 '0'，导致只查女生。
|
||
# 建议调用方在不需要筛选性别时不要传该参数，或修改此段逻辑。
|
||
if gender == '男':
|
||
gender = '1'
|
||
else:
|
||
gender = '0'
|
||
|
||
try:
|
||
# 创建基础查询
|
||
query = session.query(Student_info)
|
||
|
||
# 过滤逻辑删除：只查询未删除的记录（is_deleted = '0'）
|
||
query = query.filter(Student_info.is_deleted == '0')
|
||
|
||
# 如果传入学号，则按学号精确过滤
|
||
if student_id is not None:
|
||
query = query.filter(Student_info.student_id == student_id)
|
||
|
||
# 如果传入姓名，则按姓名精确过滤
|
||
# 若需模糊查询，可将 == 改为 like(f"%{student_name}%")
|
||
if student_name is not None:
|
||
query = query.filter(Student_info.student_name == student_name)
|
||
|
||
# 如果传入性别，则按性别过滤（已转为 '0'/'1'，再转为枚举）
|
||
if gender is not None:
|
||
# 把字符串转成枚举，SQLAlchemy 才能正确比较
|
||
query = query.filter(Student_info.gender == GenderEnum(gender))
|
||
|
||
# 返回所有匹配记录
|
||
return query.all()
|
||
except Exception as e:
|
||
# 出现异常直接向上抛出，由 API 层统一处理
|
||
raise e
|
||
|
||
|
||
# ==================== 添加学生信息 ====================
|
||
def add_student_info(session: Session, data: dict):
|
||
"""
|
||
新增一条学生记录。
|
||
:param session: 数据库会话
|
||
:param data: 学生信息字典，键需与模型字段对应
|
||
:return: 成功时返回字符串 '添加成功'
|
||
"""
|
||
try:
|
||
# 先检查学号是否已存在
|
||
rows = session.query(Student_info).filter(
|
||
Student_info.student_id == data['student_id']
|
||
).first()
|
||
|
||
# 如果已存在，返回 ValueError 类（注意：这里返回的是类，不是实例，也不是抛出异常）
|
||
# 调用方如果做 `if not res` 判断，ValueError 类是真值，可能导致逻辑错误。
|
||
# 建议改为 `raise ValueError("学号已存在")`
|
||
if rows is not None:
|
||
return ValueError
|
||
|
||
# 根据字典创建 ORM 对象
|
||
obj = Student_info(**data)
|
||
session.add(obj) # 加入会话
|
||
session.commit() # 提交事务，写入数据库
|
||
return '添加成功'
|
||
except Exception as e:
|
||
# 出现异常时回滚事务，避免脏数据
|
||
session.rollback()
|
||
raise e
|
||
|
||
|
||
# ==================== 修改学生信息 ====================
|
||
def update_student_info(session: Session, data: dict):
|
||
"""
|
||
根据学号更新学生信息（全量更新）。
|
||
:param session: 数据库会话
|
||
:param data: 包含 student_id 及待更新字段的字典
|
||
:return: 成功返回 '修改成功'；未找到记录返回 None
|
||
"""
|
||
try:
|
||
# 使用 query.update() 批量更新，返回受影响行数
|
||
rows = session.query(Student_info).filter(
|
||
Student_info.student_id == data['student_id']
|
||
).update(data)
|
||
|
||
# 如果受影响行数为 0，说明学号不存在，无记录可更新
|
||
if rows == 0:
|
||
return None
|
||
|
||
session.commit()
|
||
return '修改成功'
|
||
|
||
except IntegrityError as e:
|
||
# 捕获数据库完整性错误（外键、唯一约束等），回滚并翻译为友好提示
|
||
session.rollback()
|
||
err_msg = str(e.orig) # 获取底层 MySQL 错误信息
|
||
|
||
if 'fk_student_region' in err_msg:
|
||
# 外键 fk_student_region 失败：region_id 在地区表中不存在
|
||
raise ValueError(
|
||
f"region_id '{data.get('region_id')}' 在地区表中不存在，请先维护 region_dimension"
|
||
)
|
||
elif 'id_card' in err_msg:
|
||
# 身份证号唯一约束冲突
|
||
raise ValueError(f"身份证号 '{data.get('id_card')}' 已被其他学生占用")
|
||
else:
|
||
# 其他完整性错误
|
||
raise ValueError(f"数据完整性错误: {err_msg}")
|
||
|
||
except Exception as e:
|
||
# 其他异常同样回滚并抛出
|
||
session.rollback()
|
||
raise e
|
||
|
||
|
||
# ==================== 删除学生信息（逻辑删除） ====================
|
||
def delete_student_info(session: Session, student_id: str):
|
||
"""
|
||
根据学号进行逻辑删除（将 is_deleted 置为 '1'）。
|
||
:param session: 数据库会话
|
||
:param student_id: 学号
|
||
:return: 成功返回 '修改成功'；未找到记录返回 None
|
||
"""
|
||
try:
|
||
# 使用 update 将 is_deleted 改为 '1'，实现逻辑删除
|
||
rows = session.query(Student_info).filter(
|
||
Student_info.student_id == student_id
|
||
).update({'is_deleted': '1'})
|
||
|
||
# 没有匹配记录，返回 None
|
||
if rows == 0:
|
||
return None
|
||
|
||
session.commit()
|
||
return '修改成功'
|
||
except Exception as e:
|
||
# 异常时回滚
|
||
session.rollback()
|
||
raise e |