91 lines
3.4 KiB
Python
91 lines
3.4 KiB
Python
# dao/scores_dao.py
|
|
# 成绩表的数据访问层（录入、修改、删除、查询 + 60分红线预警）
|
|
from typing import List, Optional, Tuple
|
|
|
|
from sqlalchemy.orm import Session
|
|
|
|
from model.scores import Score
|
|
from model.students import Student
|
|
from scheme.scores import ScoreAdd
|
|
|
|
|
|
class ScoreDAO:
|
|
WARNING_LINE = 60 # 成绩红线
|
|
|
|
@staticmethod
|
|
def get_one(db: Session, stu_id: int, exam_id: int, include_deleted: bool = False) -> Optional[Score]:
|
|
"""获取指定学生某次考核的成绩（复合主键查询）"""
|
|
query = db.query(Score).filter(Score.stu_id == stu_id, Score.exam_id == exam_id)
|
|
if not include_deleted:
|
|
query = query.filter(Score.is_deleted == 0)
|
|
return query.first()
|
|
|
|
@staticmethod
|
|
def add_score(db: Session, score_data: ScoreAdd) -> Score:
|
|
"""录入成绩（调用方需先校验学生存在与复合主键冲突）"""
|
|
db_score = Score(**score_data.model_dump(), is_deleted=0)
|
|
db.add(db_score)
|
|
db.commit()
|
|
db.refresh(db_score)
|
|
return db_score
|
|
|
|
@staticmethod
|
|
def update_score(db: Session, stu_id: int, exam_id: int, new_score: float) -> Optional[Score]:
|
|
"""修改指定学生的某次成绩"""
|
|
db_score = ScoreDAO.get_one(db, stu_id, exam_id)
|
|
if db_score is None:
|
|
return None
|
|
db_score.score = new_score
|
|
db_score.is_deleted = 0 # 若曾被软删除，修改视为恢复
|
|
db.commit()
|
|
db.refresh(db_score)
|
|
return db_score
|
|
|
|
@staticmethod
|
|
def delete_score(db: Session, stu_id: int, exam_id: int) -> bool:
|
|
"""
|
|
逻辑删除指定学生的某次成绩
|
|
:return: True 成功 / False 不存在或已删除
|
|
"""
|
|
db_score = ScoreDAO.get_one(db, stu_id, exam_id)
|
|
if db_score is None:
|
|
return False
|
|
db_score.is_deleted = 1
|
|
db.commit()
|
|
return True
|
|
|
|
@staticmethod
|
|
def get_by_student(db: Session, stu_id: int) -> List[Score]:
|
|
"""获取学生的全部成绩（按考核序次排序）"""
|
|
return (
|
|
db.query(Score)
|
|
.filter(Score.stu_id == stu_id, Score.is_deleted == 0)
|
|
.order_by(Score.exam_id)
|
|
.all()
|
|
)
|
|
|
|
@staticmethod
|
|
def get_by_exam(db: Session, exam_id: int, skip: int = 0, limit: int = 100) -> Tuple[int, List[Score]]:
|
|
"""按考核序次查询成绩列表（分页）"""
|
|
query = db.query(Score).filter(Score.exam_id == exam_id, Score.is_deleted == 0)
|
|
total = query.count()
|
|
items = query.order_by(Score.stu_id).offset(skip).limit(limit).all()
|
|
return total, items
|
|
|
|
@staticmethod
|
|
def attach_student_name(db: Session, score: Score) -> None:
|
|
"""把学生姓名附加到 Score 对象上（供响应模型冗余展示）"""
|
|
student = db.query(Student).filter(Student.stu_id == score.stu_id).first()
|
|
score.stu_name = student.stu_name if student else None
|
|
|
|
@staticmethod
|
|
def build_score_item(score: Score, stu_name: Optional[str] = None) -> dict:
|
|
"""组装成绩响应字典：附带红线预警标记"""
|
|
return {
|
|
"stu_id": score.stu_id,
|
|
"exam_id": score.exam_id,
|
|
"score": score.score,
|
|
"stu_name": stu_name,
|
|
"is_warning": score.score is not None and score.score < ScoreDAO.WARNING_LINE,
|
|
}
|