101 lines
3.1 KiB
Python
101 lines
3.1 KiB
Python
from sqlalchemy.orm import Session
|
|
from sqlalchemy import select, delete
|
|
from model.wl_score_model import Score
|
|
from model.wl_student_model import Student
|
|
|
|
|
|
def get_student_by_no(db: Session, stu_no: str):
|
|
"""按学号查学生。
|
|
新增成绩前先确认学生真实存在：wl_score.stu_no 是指向学生表的外键，
|
|
给一个不存在的学号会在 commit 时抛外键异常，那样报出来就是 500 了。
|
|
"""
|
|
return db.scalar(select(Student).where(Student.stu_no == stu_no))
|
|
|
|
|
|
def get_student_by_id(db: Session, stu_id: int):
|
|
"""检测学生是否存在：通过学生id查学生表"""
|
|
return db.scalar(select(Student).where(Student.stu_id == stu_id))
|
|
|
|
|
|
def get_student_by_name(db: Session, stu_name: str):
|
|
"""通过学生姓名查询学生"""
|
|
return db.scalar(select(Student).where(Student.stu_name == stu_name))
|
|
|
|
|
|
def get_score(db: Session, stu_no: str, exam_order: int):
|
|
"""根据学号+考试批次查询单条成绩"""
|
|
stmt = select(Score).where(
|
|
Score.stu_no == stu_no,
|
|
Score.exam_order == exam_order
|
|
)
|
|
return db.scalar(stmt)
|
|
|
|
|
|
def list_score_by_stu_no(db: Session, stu_no: str):
|
|
"""查询一个学生全部成绩。
|
|
学号不存在时自然就是空列表，不用单独去查一次学生。
|
|
"""
|
|
stmt = select(Score).where(Score.stu_no == stu_no)
|
|
return db.scalars(stmt).all()
|
|
|
|
|
|
def create_score(db: Session, stu_no: str, exam_order: int, score: float):
|
|
"""
|
|
新增成绩
|
|
返回：成功返回Score对象；检测不通过返回None
|
|
"""
|
|
# 1.检测：学生是否真实存在学生表
|
|
if not get_student_by_no(db, stu_no):
|
|
print("[检测] 学生表无此学生,stu_no=", stu_no)
|
|
return None
|
|
|
|
# 2.检测：该学生该批次成绩是否已经存在（重复新增）
|
|
if get_score(db, stu_no, exam_order):
|
|
print("[检测] 成绩已存在 stu_no=", stu_no, "exam_order=", exam_order)
|
|
return None
|
|
|
|
# 3.构建ORM对象插入数据库
|
|
new_score = Score(
|
|
stu_no=stu_no,
|
|
exam_order=exam_order,
|
|
score=score
|
|
)
|
|
db.add(new_score)
|
|
db.commit()
|
|
db.refresh(new_score)
|
|
return new_score
|
|
|
|
|
|
def update_score(db: Session, stu_no: str, exam_order: int, new_score: float):
|
|
"""
|
|
更新指定学生指定考试批次的分数
|
|
"""
|
|
# 检测这条成绩记录是否存在（记录在，学生就一定在，外键保证的）
|
|
score_obj = get_score(db, stu_no, exam_order)
|
|
if not score_obj:
|
|
print("[检测] 更新失败：该学生该批次成绩记录不存在")
|
|
return None
|
|
|
|
# 直接修改对象属性
|
|
score_obj.score = new_score
|
|
db.commit()
|
|
db.refresh(score_obj)
|
|
return score_obj
|
|
|
|
|
|
def delete_score(db: Session, stu_no: str, exam_order: int):
|
|
"""删除某个学生某一次考核成绩"""
|
|
# 检测记录是否存在
|
|
score_obj = get_score(db, stu_no, exam_order)
|
|
if not score_obj:
|
|
print("[检测] 删除失败，记录不存在")
|
|
return False
|
|
|
|
stmt = delete(Score).where(
|
|
Score.stu_no == stu_no,
|
|
Score.exam_order == exam_order
|
|
)
|
|
db.execute(stmt)
|
|
db.commit()
|
|
return True
|