77 lines
2.7 KiB
Python
77 lines
2.7 KiB
Python
# dao/classes_dao.py
|
|
# 班级表的数据访问层（增、查、改、逻辑删除）
|
|
from typing import Tuple
|
|
|
|
from sqlalchemy.orm import Session
|
|
|
|
from model.classes import Classinfo
|
|
from model.students import Student
|
|
from scheme.classes import ClassAdd, ClassUpdate
|
|
|
|
|
|
class ClassDAO:
|
|
@staticmethod
|
|
def inspect_class_id_unq(db: Session, class_id: int):
|
|
"""
|
|
根据班级ID获取班级对象（用于主键唯一性检查，查询结果包含被软删除的对象）
|
|
"""
|
|
return db.query(Classinfo).filter(Classinfo.class_id == class_id).first()
|
|
|
|
@staticmethod
|
|
def get_active(db: Session, class_id: int):
|
|
"""获取未被软删除的班级对象（用于外键校验）"""
|
|
return (
|
|
db.query(Classinfo)
|
|
.filter(Classinfo.class_id == class_id, Classinfo.is_deleted == 0)
|
|
.first()
|
|
)
|
|
|
|
@staticmethod
|
|
def add_class(db: Session, class_data: ClassAdd) -> Classinfo:
|
|
"""新增班级；class_id 不传则使用自增"""
|
|
db_class = Classinfo(**class_data.model_dump(exclude_none=True))
|
|
db.add(db_class)
|
|
db.commit()
|
|
db.refresh(db_class)
|
|
return db_class
|
|
|
|
@staticmethod
|
|
def update(db: Session, class_id: int, class_data: ClassUpdate):
|
|
"""更新班级（只更新传入的非 None 字段）"""
|
|
db_class = ClassDAO.get_active(db, class_id)
|
|
if db_class is None:
|
|
return None
|
|
for key, value in class_data.model_dump(exclude_unset=True, exclude_none=True).items():
|
|
setattr(db_class, key, value)
|
|
db.commit()
|
|
db.refresh(db_class)
|
|
return db_class
|
|
|
|
@staticmethod
|
|
def delete_light(db: Session, class_id: int) -> bool:
|
|
"""
|
|
逻辑删除班级；若班级下仍有未删除学生则拒绝删除
|
|
:return: True 成功 / False 不存在 / -1 班级下还有学生
|
|
"""
|
|
db_class = ClassDAO.get_active(db, class_id)
|
|
if db_class is None:
|
|
return False
|
|
student_count = (
|
|
db.query(Student)
|
|
.filter(Student.class_id == class_id, Student.is_deleted == 0)
|
|
.count()
|
|
)
|
|
if student_count > 0:
|
|
return -1
|
|
db_class.is_deleted = 1
|
|
db.commit()
|
|
return True
|
|
|
|
@staticmethod
|
|
def get_all(db: Session, skip: int = 0, limit: int = 100) -> Tuple[int, list]:
|
|
"""分页获取所有班级（不含软删除），返回 (总数, 列表)"""
|
|
query = db.query(Classinfo).filter(Classinfo.is_deleted == 0)
|
|
total = query.count()
|
|
items = query.order_by(Classinfo.class_id).offset(skip).limit(limit).all()
|
|
return total, items
|