80 lines
2.4 KiB
Python
80 lines
2.4 KiB
Python
|
from sqlalchemy.orm import Session
|
||
|
|
|
||
|
|
from model import Score
|
||
|
|
from schema.score_schema import ScoreUpdate, ScoreCreate
|
||
|
|
|
||
|
|
|
||
|
|
#添加成绩
|
||
|
|
def create_score(
|
||
|
|
db:Session,
|
||
|
|
score_data:ScoreCreate,
|
||
|
|
):
|
||
|
|
if not isinstance(score_data.student_id, int) or score_data.student_id < 1:
|
||
|
|
raise ValueError("学生ID不合法")
|
||
|
|
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,
|
||
|
|
):
|
||
|
|
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,
|
||
|
|
):
|
||
|
|
if not isinstance(student_id, int) or student_id < 1:
|
||
|
|
raise ValueError("学生ID不合法")
|
||
|
|
if exam_id not in (1, 2, 3):
|
||
|
|
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):
|
||
|
|
if not isinstance(student_id, int) or student_id < 1:
|
||
|
|
raise ValueError("学生ID不合法")
|
||
|
|
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
|