86 lines
2.3 KiB
Python
86 lines
2.3 KiB
Python
from sqlalchemy import DECIMAL
|
|
from models.score import Score
|
|
|
|
def post_score_dao(data_dict, db):
|
|
"""新增成绩"""
|
|
try:
|
|
obj = Score(**data_dict)
|
|
db.add(obj)
|
|
except Exception as e:
|
|
print("新增成绩异常：", e)
|
|
db.rollback()
|
|
return False
|
|
else:
|
|
db.commit()
|
|
return True
|
|
|
|
|
|
def update_score_dao(score_id, db, updates):
|
|
"""修改成绩，query.update方式"""
|
|
try:
|
|
db.query(Score).filter(
|
|
Score.id == score_id,
|
|
Score.is_deleted == 0
|
|
).update(updates)
|
|
except Exception:
|
|
db.rollback()
|
|
return False
|
|
else:
|
|
db.commit()
|
|
return True
|
|
|
|
|
|
def get_score_dao(score_id, student_id, exam_seq, page, page_size, db):
|
|
"""分页查询成绩，多条件过滤，只查未删除"""
|
|
q = db.query(Score).filter(Score.is_deleted == 0)
|
|
if score_id:
|
|
q = q.filter(Score.id == score_id)
|
|
if student_id:
|
|
q = q.filter(Score.student_id == student_id)
|
|
if exam_seq:
|
|
q = q.filter(Score.exam_seq == exam_seq)
|
|
|
|
total = q.count()
|
|
rows = q.offset((page - 1) * page_size).limit(page_size).all()
|
|
|
|
res_list = [
|
|
{
|
|
"id": i.id,
|
|
"student_id": i.student_id,
|
|
"score": i.score,
|
|
"exam_seq": i.exam_seq,
|
|
"exam_date": i.exam_date,
|
|
"remark": i.remark,
|
|
"created_at": i.created_at,
|
|
"updated_at": i.updated_at,
|
|
"is_deleted": i.is_deleted
|
|
}
|
|
for i in rows
|
|
]
|
|
return total, res_list
|
|
|
|
|
|
def delete_score_dao(score_id, db):
|
|
"""逻辑删除，更新is_deleted=1"""
|
|
try:
|
|
db.query(Score).filter(
|
|
Score.id == score_id,
|
|
Score.is_deleted == 0
|
|
).update({"is_deleted": 1})
|
|
except Exception as err:
|
|
print("删除成绩异常：", err)
|
|
db.rollback()
|
|
return False
|
|
else:
|
|
db.commit()
|
|
return True
|
|
|
|
|
|
def exist_student_exam_dao(student_id, exam_seq, db):
|
|
"""校验：同一个学生+考试序次是否已经存在，用于联合唯一业务判断"""
|
|
rec = db.query(Score).filter(
|
|
Score.student_id == student_id,
|
|
Score.exam_seq == exam_seq,
|
|
Score.is_deleted == 0
|
|
).first()
|
|
return rec is not None |