62 lines
2.5 KiB
Python
62 lines
2.5 KiB
Python
# api/stu_score_api.py
|
|
# 学生成绩管理模块：接口层只负责 HTTP 出入参，等级判定与查库交给 dao.stu_score_dao
|
|
|
|
from typing import Optional
|
|
|
|
from fastapi import APIRouter, Depends, Query
|
|
from sqlalchemy.orm import Session
|
|
|
|
from dao import stu_score_dao
|
|
from database import get_db
|
|
from schemas.stu_score_schema import StuScoreCreate, StuScoreUpdate, StuScoreOut
|
|
|
|
router = APIRouter()
|
|
|
|
|
|
@router.get("", summary="分页查询成绩列表")
|
|
def list_scores(
|
|
page: int = Query(1, ge=1, description="页码"),
|
|
size: int = Query(10, ge=1, le=100, description="每页数量"),
|
|
stu_id: Optional[str] = Query(None, description="按学生编号过滤"),
|
|
score_level: Optional[str] = Query(None, description="按成绩等级过滤"),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
total, items = stu_score_dao.list_scores(
|
|
db, page=page, size=size, stu_id=stu_id, score_level=score_level
|
|
)
|
|
return {
|
|
"total": total,
|
|
"page": page,
|
|
"size": size,
|
|
"items": [StuScoreOut.model_validate(x) for x in items],
|
|
}
|
|
|
|
|
|
@router.get("/student/{stu_id}", summary="查询某学生的全部成绩")
|
|
def get_student_scores(stu_id: str, db: Session = Depends(get_db)):
|
|
scores = stu_score_dao.list_scores_by_student(db, stu_id)
|
|
return {"stu_id": stu_id, "items": [StuScoreOut.model_validate(s) for s in scores]}
|
|
|
|
|
|
@router.get("/{score_id}", response_model=StuScoreOut, summary="查询单条成绩记录")
|
|
def get_score(score_id: int, db: Session = Depends(get_db)):
|
|
return stu_score_dao.assert_score_alive(db, score_id)
|
|
|
|
|
|
@router.post("", response_model=StuScoreOut, status_code=201, summary="新增成绩记录")
|
|
def create_score(payload: StuScoreCreate, db: Session = Depends(get_db)):
|
|
"""新增成绩：未显式指定等级时，由 DAO 按分数自动判定。"""
|
|
return stu_score_dao.create_score(db, payload.model_dump())
|
|
|
|
|
|
@router.put("/{score_id}", response_model=StuScoreOut, summary="更新成绩记录")
|
|
def update_score(score_id: int, payload: StuScoreUpdate, db: Session = Depends(get_db)):
|
|
"""更新成绩：只改分数没改等级时，由 DAO 自动重判等级。"""
|
|
return stu_score_dao.update_score(db, score_id, payload.model_dump(exclude_unset=True))
|
|
|
|
|
|
@router.delete("/{score_id}", summary="删除成绩记录(软删除)")
|
|
def delete_score(score_id: int, db: Session = Depends(get_db)):
|
|
"""软删除成绩记录。"""
|
|
stu_score_dao.soft_delete_score(db, score_id)
|
|
return {"message": "删除成功", "id": score_id} |