83 lines
2.7 KiB
Python
83 lines
2.7 KiB
Python
# dao/stu_dao.py
|
|
# 学生信息数据访问层
|
|
|
|
from typing import List, Optional, Tuple
|
|
|
|
from sqlalchemy.orm import Session
|
|
|
|
from dao.advisor_dao import assert_advisor_alive
|
|
from dao.cls_mgmt_dao import assert_cls_alive
|
|
from dao.exceptions import ConflictError, NotFoundError
|
|
from dao.pagination import paginate
|
|
from model.stu_model import StuInfo
|
|
|
|
|
|
def assert_stu_alive(db: Session, stu_id: str, detail: Optional[str] = None) -> StuInfo:
|
|
"""
|
|
确认学生存在且未删除，否则抛 NotFoundError。
|
|
供成绩、就业模块复用作外键校验。
|
|
"""
|
|
stu = db.query(StuInfo).filter(StuInfo.id == stu_id, StuInfo.is_deleted == 0).first()
|
|
if not stu:
|
|
raise NotFoundError(detail or "学生不存在")
|
|
return stu
|
|
|
|
|
|
def list_students(
|
|
db: Session,
|
|
page: int,
|
|
size: int,
|
|
name: Optional[str] = None,
|
|
cls_id: Optional[str] = None,
|
|
state: Optional[str] = None,
|
|
) -> Tuple[int, List[StuInfo]]:
|
|
"""分页查询未删除的学生，支持按姓名模糊 / 班级 / 状态过滤。"""
|
|
query = db.query(StuInfo).filter(StuInfo.is_deleted == 0)
|
|
if name:
|
|
query = query.filter(StuInfo.name.like(f"%{name}%"))
|
|
if cls_id:
|
|
query = query.filter(StuInfo.cls_id == cls_id)
|
|
if state:
|
|
query = query.filter(StuInfo.state == state)
|
|
return paginate(query, page, size, order_by=StuInfo.id)
|
|
|
|
|
|
def create_student(db: Session, data: dict) -> StuInfo:
|
|
"""新增学生：校验学号唯一，且所属班级、顾问（若填写）都存在。"""
|
|
# 学号是主键，已软删除的记录同样占用该学号，所以这里不加 is_deleted 过滤
|
|
if db.query(StuInfo).filter(StuInfo.id == data["id"]).first():
|
|
raise ConflictError("学生编号已存在")
|
|
|
|
assert_cls_alive(db, data["cls_id"], "所属班级不存在")
|
|
if data.get("advisor_id"):
|
|
assert_advisor_alive(db, data["advisor_id"], "顾问不存在")
|
|
|
|
stu = StuInfo(**data)
|
|
db.add(stu)
|
|
db.commit()
|
|
db.refresh(stu)
|
|
return stu
|
|
|
|
|
|
def update_student(db: Session, stu_id: str, data: dict) -> StuInfo:
|
|
"""更新学生：若更换班级 / 顾问需重新校验外键。"""
|
|
stu = assert_stu_alive(db, stu_id, "学生不存在或已删除")
|
|
|
|
if data.get("cls_id"):
|
|
assert_cls_alive(db, data["cls_id"], "所属班级不存在")
|
|
if data.get("advisor_id"):
|
|
assert_advisor_alive(db, data["advisor_id"], "顾问不存在")
|
|
|
|
for field, value in data.items():
|
|
setattr(stu, field, value)
|
|
|
|
db.commit()
|
|
db.refresh(stu)
|
|
return stu
|
|
|
|
|
|
def soft_delete_student(db: Session, stu_id: str) -> None:
|
|
"""软删除学生。"""
|
|
stu = assert_stu_alive(db, stu_id, "学生不存在或已删除")
|
|
stu.is_deleted = 1
|
|
db.commit() |