api,dao,schemas已完成

api/score.py

@@ -0,0 +1,51 @@
+from fastapi import APIRouter, Depends, HTTPException, Query
+from sqlalchemy.orm import Session
+from core.database import get_db
+from dao.score import post_score_dao, update_score_dao, get_score_dao, delete_score_dao, exist_student_exam_dao
+from schemas.score import ScoreCreate, ScoreUpdate, ScorePageResp
+
+scores_router = APIRouter(prefix="/scores", tags=["成绩接口"])
+
+
+@scores_router.post("", summary="新增成绩接口")
+def create_score(req: ScoreCreate, db: Session = Depends(get_db)):
+    # 业务校验：联合唯一
+    if exist_student_exam_dao(req.student_id, req.exam_seq, db):
+        raise HTTPException(status_code=400, detail="该学生本次考试成绩已存在，不能重复录入")
+
+    ok = post_score_dao(req.model_dump(), db)
+    if not ok:
+        raise HTTPException(status_code=400, detail="新增成绩失败，请检查学生id是否存在")
+    return {"msg": "新增成功"}
+
+
+@scores_router.get("", response_model=ScorePageResp, summary="查询成绩接口")
+def list_score(
+    score_id: int|None=None,
+    student_id: int |None=None,
+    exam_seq: int|None=None,
+    page: int = Query(1, ge=1),
+    page_size: int = Query(10, ge=1, le=100),
+    db: Session = Depends(get_db)
+):
+    total, data_list = get_score_dao(score_id, student_id, exam_seq, page, page_size, db)
+    return {"total": total, "data": data_list}
+
+
+@scores_router.put("{score_id}", summary="更新成绩接口")
+def modify_score(score_id: int, req: ScoreUpdate, db: Session = Depends(get_db)):
+    update_dict = req.model_dump(exclude_unset=True)
+    if not update_dict:
+        raise HTTPException(status_code=400, detail="没有传入任何修改字段")
+    ok = update_score_dao(score_id, db, update_dict)
+    if not ok:
+        raise HTTPException(status_code=400, detail="修改失败，记录不存在或已删除")
+    return {"msg": "修改成功"}
+
+
+@scores_router.delete("{score_id}", summary="删除成绩接口")
+def remove_score(score_id: int, db: Session = Depends(get_db)):
+    ok = delete_score_dao(score_id, db)
+    if not ok:
+        raise HTTPException(status_code=400, detail="删除失败，记录不存在或已删除")
+    return {"msg": "逻辑删除成功"}

dao/score.py

@@ -0,0 +1,86 @@
+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

schemas/score.py

@@ -0,0 +1,36 @@
+from pydantic import BaseModel, Field
+from datetime import date, datetime
+from typing import Optional, List
+from decimal import Decimal
+
+
+class ScoreCreate(BaseModel):
+    student_id: int
+    exam_seq: int
+    score: Decimal = Field(ge=0, le=100, description="分数0‑100")
+    exam_date: Optional[date] = None
+    remark: Optional[str] = None
+
+
+class ScoreUpdate(BaseModel):
+    score: Optional[Decimal] = Field(None, ge=0, le=100)
+    exam_seq: Optional[int] = None
+    exam_date: Optional[date] = None
+    remark: Optional[str] = None
+
+
+class ScoreItem(BaseModel):
+    id: int
+    student_id: int
+    score: Decimal
+    exam_seq: int
+    exam_date: Optional[date]
+    remark: Optional[str]
+    created_at: Optional[datetime]
+    updated_at: Optional[datetime]
+    is_deleted: int
+
+
+class ScorePageResp(BaseModel):
+    total: int
+    data: List[ScoreItem]