96 lines
2.9 KiB
Python
96 lines
2.9 KiB
Python
from sqlalchemy.orm import Session
|
|
|
|
from model import Score, Student
|
|
from schema.score_schema import ScoreUpdate, ScoreCreate
|
|
|
|
|
|
#添加成绩
|
|
def create_score(
|
|
db:Session,
|
|
score_data:ScoreCreate,
|
|
):
|
|
stu = db.query(Student).filter(Student.sid == score_data.student_id, Student.flag == 1).first()
|
|
if stu is None:
|
|
raise ValueError("该学生不存在")
|
|
if score_data.exam_id not in (1, 2, 3):
|
|
raise ValueError("考试次序不合法")
|
|
if not isinstance(score_data.score, int) or not (0 <= score_data.score <= 100):
|
|
raise ValueError("成绩不合法")
|
|
exist = db.query(Score).filter(Score.student_id == score_data.student_id,Score.exam_id == score_data.exam_id,Score.flag == 1).first()
|
|
if exist:
|
|
return None
|
|
new_score = Score(
|
|
student_id=score_data.student_id,
|
|
exam_id=score_data.exam_id,
|
|
score=score_data.score,
|
|
flag=1
|
|
)
|
|
#添加到数据库会话
|
|
db.add(new_score)
|
|
db.commit()
|
|
#覆盖Python原始数据
|
|
db.refresh(new_score)
|
|
return new_score
|
|
|
|
#查询学生成绩
|
|
def search_score(
|
|
db:Session,
|
|
student_id:int,
|
|
exam_id:int,
|
|
):
|
|
stu = db.query(Student).filter(Student.sid == student_id, Student.flag == 1).first()
|
|
if stu is None:
|
|
raise ValueError("该学生不存在")
|
|
if exam_id not in (1, 2, 3):
|
|
raise ValueError("考试次序不合法")
|
|
return db.query(Score).filter(
|
|
Score.flag == 1,
|
|
Score.student_id == student_id,
|
|
Score.exam_id == exam_id
|
|
).all()
|
|
|
|
#修改成绩
|
|
def update_score(
|
|
db:Session,
|
|
student_id:int,
|
|
exam_id:int,
|
|
score_data:ScoreUpdate,
|
|
):
|
|
stu = db.query(Student).filter(Student.sid == student_id, Student.flag == 1).first()
|
|
if stu is None:
|
|
raise ValueError("该学生不存在")
|
|
if exam_id not in (1, 2, 3):
|
|
raise ValueError("考试次序不合法")
|
|
if not isinstance(score_data.score, int) or not (0 <= score_data.score <= 100):
|
|
raise ValueError("成绩不合法")
|
|
score = db.query(Score).filter(Score.student_id == student_id,Score.exam_id == exam_id,Score.flag == 1).first()
|
|
if score is None:
|
|
return None
|
|
update_data = score_data.model_dump(exclude_unset=True)
|
|
for k,v in update_data.items():
|
|
setattr(score,k,v)
|
|
db.commit()
|
|
db.refresh(score)
|
|
return score
|
|
|
|
#删除成绩
|
|
def delete_score(db:Session,student_id:int,exam_id:int):
|
|
stu = db.query(Student).filter(Student.sid == student_id, Student.flag == 1).first()
|
|
if stu is None:
|
|
raise ValueError("该学生不存在")
|
|
|
|
if exam_id not in (1, 2, 3):
|
|
raise ValueError("考试次序不合法")
|
|
|
|
score = db.query(Score).filter(
|
|
Score.student_id == student_id,
|
|
Score.flag == 1,
|
|
Score.exam_id == exam_id
|
|
).first()
|
|
if score is None:
|
|
return None
|
|
#逻辑删除，磁盘保留
|
|
score.flag = 0
|
|
db.commit()
|
|
db.refresh(score)
|
|
return score |