60 lines
2.2 KiB
Python
60 lines
2.2 KiB
Python
# dao/score_dao.py
|
|
# 本文件封装对学生成绩表的所有数据库操作（增、删、改、查）
|
|
from fastapi import HTTPException
|
|
from sqlalchemy import and_
|
|
from sqlalchemy.orm import Session
|
|
|
|
from model.student_score import Score
|
|
from scheme.scores import ScoreCreate, ScoreUpdate
|
|
|
|
|
|
class ScoreDAO:
|
|
|
|
# 根据学生id、考试次数查询有效成绩（flag=1）
|
|
@staticmethod
|
|
def get_by_id(db: Session, sid: int, exam_num: int):
|
|
score = db.query(Score).filter(and_(Score.sid == sid, Score.exam_num == exam_num, Score.flag == 1)
|
|
).first()
|
|
if score:
|
|
return score
|
|
raise HTTPException(status_code=404, detail="学生分数不存在")
|
|
|
|
# 查询某个学生的所有考试成绩
|
|
@staticmethod
|
|
def get_by_student(db: Session, sid: int):
|
|
return (db.query(Score)
|
|
.filter(Score.sid == sid, Score.flag == 1).order_by(Score.exam_num).all())
|
|
|
|
# 录入成绩
|
|
@staticmethod
|
|
def create(db: Session, data: ScoreCreate):
|
|
# 同一个学生同一次考试只能有一条有效成绩
|
|
exist = db.query(Score).filter(and_(Score.sid == data.sid, Score.exam_num == data.exam_num, Score.flag == 1)).first()
|
|
if exist:
|
|
raise HTTPException(status_code=400,
|
|
detail=f"学生 {data.sid} 第 {data.exam_num} 次成绩已存在，请使用修改接口")
|
|
score = Score(sid=data.sid, exam_num=data.exam_num, score=data.score, flag=1)
|
|
db.add(score)
|
|
db.commit()
|
|
db.refresh(score)
|
|
return score
|
|
|
|
# 根据学生id和考试次数更新成绩
|
|
@staticmethod
|
|
def update(db: Session, sid: int, exam_num: int, score_data: ScoreUpdate):
|
|
update_data = score_data.model_dump()
|
|
db.query(Score) \
|
|
.filter(Score.sid == sid) \
|
|
.update(update_data)
|
|
db.commit() # 提交事务
|
|
return ScoreDAO.get_by_id(db, sid, exam_num)
|
|
|
|
# 根据学生id和考试次数逻辑删除（flag 改为 0）
|
|
@staticmethod
|
|
def delete(db: Session, sid: int, exam_num: int):
|
|
db_score = ScoreDAO.get_by_id(db, sid, exam_num)
|
|
db_score.flag = 0 # 标记为已删除
|
|
db.commit()
|
|
db.refresh(db_score)
|
|
return db_score
|