# dao/students_dao.py # 学生表的数据访问层(增、查、改、逻辑删除 + 学号自动生成规则) from datetime import date from typing import Optional, Tuple from sqlalchemy import or_ from sqlalchemy.orm import Session, joinedload from model.students import Student from model.classes import Classinfo from model.advisor import Advisor from scheme.students import StudentAdd, StudentUpdate class StudentDAO: @staticmethod def inspect_student_id_unq(db: Session, stu_id: int) -> Optional[Student]: """根据学号获取学生对象(用于主键唯一性检查,包含被软删除的对象)""" return db.query(Student).filter(Student.stu_id == stu_id).first() @staticmethod def get_active(db: Session, stu_id: int) -> Optional[Student]: """获取未被软删除的学生对象(用于外键校验与查询)""" return ( db.query(Student) .filter(Student.stu_id == stu_id, Student.is_deleted == 0) .first() ) @staticmethod def inspect_class_id_unq(db: Session, class_id: int) -> Optional[Classinfo]: """班级外键校验(排除软删除)""" return ( db.query(Classinfo) .filter(Classinfo.class_id == class_id, Classinfo.is_deleted == 0) .first() ) @staticmethod def inspect_advisor_id_unq(db: Session, advisor_id: int) -> Optional[Advisor]: """顾问外键校验(排除软删除)""" return ( db.query(Advisor) .filter(Advisor.advisor_id == advisor_id, Advisor.is_deleted == 0) .first() ) # ==================== 学号生成规则 ==================== # 规则:入学年份(4位) + 班级号(2位) + 班内序号(4位) # 示例:2026 年入学、班级 3 的第 1 个学生 -> 2026030001 # 说明:班级号超过 99 时会溢出到序号位,业务上建议班级数控制在 99 以内 @staticmethod def generate_stu_id(db: Session, enroll_time: date, class_id: int) -> int: prefix = enroll_time.year * 10000 + class_id # 前6位:年份+班级 # 找到同一前缀下最大的学号,在其基础上 +1 max_stu_id = ( db.query(Student.stu_id) .filter( Student.stu_id >= prefix * 10000, Student.stu_id <= prefix * 10000 + 9999, ) .order_by(Student.stu_id.desc()) .first() ) seq = (max_stu_id[0] % 10000) + 1 if max_stu_id else 1 return prefix * 10000 + seq # ==================== CRUD ==================== @staticmethod def add_student(db: Session, student_data: StudentAdd) -> Student: """新增学生;stu_id 不传则按规则自动生成""" data = student_data.model_dump(exclude_none=True) data.pop("is_deleted", None) if "stu_id" not in data: data["stu_id"] = StudentDAO.generate_stu_id( db, data["enroll_time"], data["class_id"] ) db_student = Student(**data) db.add(db_student) db.commit() db.refresh(db_student) return db_student @staticmethod def update(db: Session, stu_id: int, student_data: StudentUpdate) -> Optional[Student]: """更新学生信息(只更新传入的非 None 字段)""" db_student = StudentDAO.get_active(db, stu_id) if db_student is None: return None for key, value in student_data.model_dump(exclude_unset=True, exclude_none=True).items(): setattr(db_student, key, value) db.commit() db.refresh(db_student) return db_student @staticmethod def delete_light(db: Session, stu_id: int) -> bool: """ 逻辑删除学生 :return: True 成功 / False 学号不存在或已被软删除 """ db_student = StudentDAO.get_active(db, stu_id) if db_student is None: return False db_student.is_deleted = 1 db.commit() return True @staticmethod def get_all( db: Session, skip: int = 0, limit: int = 100, stu_id: Optional[int] = None, stu_name: Optional[str] = None, class_id: Optional[int] = None, gender: Optional[str] = None, status: Optional[str] = None, education: Optional[str] = None, ) -> Tuple[int, list]: """ 分页 + 多条件筛选查询学生列表(不含软删除) 支持按编号精确、姓名模糊、班级、性别、状态、学历筛选 """ query = ( db.query(Student) .options(joinedload(Student.classes), joinedload(Student.advisor)) .filter(Student.is_deleted == 0) ) if stu_id is not None: query = query.filter(Student.stu_id == stu_id) if stu_name: query = query.filter(Student.stu_name.like(f"%{stu_name}%")) if class_id is not None: query = query.filter(Student.class_id == class_id) if gender: query = query.filter(Student.gender == gender) if status: query = query.filter(Student.status == status) if education: query = query.filter(Student.education == education) total = query.count() items = ( query.order_by(Student.stu_id) .offset(skip) .limit(limit) .all() ) return total, items