106 lines
3.0 KiB
Python
106 lines
3.0 KiB
Python
from models.score import Score
|
|
|
|
def post_score_dao(data_dict, db):
|
|
"""新增成绩"""
|
|
try:
|
|
obj = Score(**data_dict)
|
|
db.add(obj)
|
|
db.commit()
|
|
except Exception as e:
|
|
print("新增成绩异常：", e)
|
|
db.rollback()
|
|
return False
|
|
else:
|
|
return True
|
|
|
|
|
|
def update_score_dao(score_id, db, updates):
|
|
"""修改成绩，query.update方式"""
|
|
try:
|
|
cnt = db.query(Score).filter(
|
|
Score.id == score_id,
|
|
Score.is_deleted == 0
|
|
).update(updates)
|
|
if cnt == 0:
|
|
db.rollback()
|
|
return False
|
|
except Exception as e:
|
|
print("修改成绩异常：", e)
|
|
db.rollback()
|
|
return False
|
|
else:
|
|
db.commit()
|
|
return True
|
|
|
|
|
|
def get_score_dao(score_id, student_id, exam_seq, page, page_size, db):
|
|
"""分页查询成绩，多条件过滤，只查未删除
|
|
规则：
|
|
- score_id == 0 → 查所有学生成绩（仅分页，不加 id 过滤）
|
|
- 三个筛选条件全为空（None） → 直接返回空，不查库
|
|
- 其他情况 → 按传入条件过滤
|
|
"""
|
|
q = db.query(Score).filter(Score.is_deleted == 0)
|
|
|
|
if score_id == 0:
|
|
pass # score_id=0 → 查全部（分页）
|
|
elif score_id is None and not student_id and not exam_seq:
|
|
return 0, [] # 三个参数都没传 → 返回空
|
|
else:
|
|
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):
|
|
try:
|
|
cnt = db.query(Score).filter(
|
|
Score.id == score_id,
|
|
Score.is_deleted == 0
|
|
).update({"is_deleted": 1}) # ← 接住返回值
|
|
if cnt == 0: # ← 关键：一行都没匹配到
|
|
db.rollback()
|
|
return False
|
|
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 |