增添查询所有成绩接口，dao层运行代码-曾凯

Reviewed-on: #36

api/employment.py

@@ -1,19 +1,16 @@
 from fastapi import APIRouter, Depends, HTTPException, Query
 from sqlalchemy.orm import Session
 from core.database import get_db
-from dao.employment import (
-    create_employment,
-    get_employment_by_id,
-    get_employment_by_student_id,
-    get_employment_list,
-    update_employment,
-    delete_employment_logic
-)
-from schemas.employment import (
-    EmploymentCreate,
-    EmploymentUpdate,
-    EmploymentResponse
-)
+from dao.employment import (create_employment,
+                            get_employment_by_id,
+                            get_employment_by_student_id,
+                            get_employment_list,
+                            update_employment,
+                            delete_employment_logic)
+
+from schemas.employment import (EmploymentCreate,
+                                EmploymentUpdate,
+                                EmploymentResponse)
 
 employment_router = APIRouter(prefix="/employments", tags=["就业管理模块"])
 

api/score.py

@@ -1,7 +1,7 @@
 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 dao.score import post_score_dao, update_score_dao, get_score_dao, get_all_scores_dao,delete_score_dao, exist_student_exam_dao
 from schemas.score import ScoreCreate, ScoreUpdate, ScorePageResp
 
 scores_router = APIRouter(prefix="/scores", tags=["成绩接口"])
@@ -12,14 +12,13 @@ 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="查询成绩接口")
+@scores_router.get("", response_model=ScorePageResp, summary="个人成绩查询接口")
 def list_score(
     score_id: int|None=None,
     student_id: int |None=None,
@@ -32,6 +31,16 @@ def list_score(
     return {"total": total, "data": data_list}
 
 
+
+@scores_router.get("/all", response_model=ScorePageResp,
+                   summary="所有学生成绩接口",
+                  )
+def list_all_scores(page: int = Query(1, ge=1),
+                    page_size: int = Query(10, ge=1, le=100),
+                    db: Session = Depends(get_db)):
+    total, data_list = get_all_scores_dao(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)

dao/employment.py

@@ -6,31 +6,22 @@ from schemas.employment import EmploymentCreate, EmploymentUpdate, EmploymentRes
 def create_employment(db, obj_in):
     """新增就业信息：一个学生只能存在一条就业记录"""
     try:
-        # obj_in.model_dump()把pydantic请求对象转为字典，解包传给Employment生成ORM对象
         db_obj = Employment(**obj_in.model_dump())
-        # add：把对象加入数据库会话，此时还没写入数据库
-        db.add(db_obj)
-        # commit：提交事务，真正写入mysql数据库
-        db.commit()
-        # refresh：从数据库刷新对象，拿到数据库自动生成的id、默认时间等字段
-        db.refresh(db_obj)
-        # model_validate：把sqlalchemy ORM对象解析进pydantic响应模型
-        # model_dump：把pydantic模型转为普通python字典返回给上层api
+        db.add(db_obj)                      # add：把对象加入数据库会话，此时还没写入数据库
+        db.commit()                         # commit：提交事务，真正写入mysql数据库
+        db.refresh(db_obj)                  # refresh：从数据库刷新对象，拿到数据库自动生成的id、默认时间等字段
         return EmploymentResponse.model_validate(db_obj).model_dump()
+        # model_validate：把sqlalchemy ORM对象解析进pydantic响应模型
     except Exception:
-        # 发生任何异常，执行回滚，撤销本次会话中未提交的数据库改动
         db.rollback()
-        # 异常情况返回None，上层api可以判断新增失败
         return None
 
 
 def get_employment_by_id(db, eid):
     """根据主键id查询就业详情，过滤逻辑删除"""
     try:
-        db_obj = db.query(Employment).filter(
-            Employment.id == eid,
-            Employment.is_deleted == 0
-        ).first()
+        db_obj = db.query(Employment).filter(Employment.id == eid
+                                             ,Employment.is_deleted == 0).first()
         if not db_obj:
             return None
         return EmploymentResponse.model_validate(db_obj).model_dump()
@@ -41,10 +32,8 @@ def get_employment_by_id(db, eid):
 def get_employment_by_student_id(db, student_id):
     """根据学生id查询就业信息，过滤逻辑删除"""
     try:
-        db_obj = db.query(Employment).filter(
-            Employment.student_id == student_id,
-            Employment.is_deleted == 0
-        ).first()
+        db_obj = db.query(Employment).filter(Employment.student_id == student_id
+                                             ,Employment.is_deleted == 0).first()
         if not db_obj:
             return None
         return EmploymentResponse.model_validate(db_obj).model_dump()
@@ -78,16 +67,11 @@ def get_employment_list(db, company_name=None, salary_min=None, salary_max=None,
 def update_employment(db, eid, obj_in):
     """修改就业信息，只更新传入的字段"""
     try:
-        # 根据id查询未删除的记录
-        db_obj_raw = db.query(Employment).filter(
-            Employment.id == eid,
-            Employment.is_deleted == 0
-        ).first()
-        # 记录不存在直接返回None
+        db_obj_raw = db.query(Employment).filter(Employment.id == eid
+                                                ,Employment.is_deleted == 0).first()
         if db_obj_raw is None:
             return None
-        # exclude_unset=True：只取出前端实际传过来的字段，不会带上没传的None字段
-        update_dict = obj_in.model_dump(exclude_unset=True)
+        update_dict = obj_in.model_dump(exclude_unset=True)# exclude_unset=True：只取出前端实际传过来的字段，不会带上没传的None字段
         for key, value in update_dict.items():
             setattr(db_obj_raw, key, value)
         db.commit()
@@ -101,10 +85,8 @@ def update_employment(db, eid, obj_in):
 def delete_employment_logic(db, eid):
     """逻辑删除，只修改is_deleted=1，不做物理删除"""
     try:
-        db_obj_raw = db.query(Employment).filter(
-            Employment.id == eid,
-            Employment.is_deleted == 0
-        ).first()
+        db_obj_raw = db.query(Employment).filter(Employment.id == eid
+                                                 ,Employment.is_deleted == 0).first()
         if db_obj_raw is None:
             return False
         db_obj_raw.is_deleted = 1

dao/score.py

@@ -1,4 +1,3 @@
-from sqlalchemy import DECIMAL
 from models.score import Score
 
 def post_score_dao(data_dict, db):
@@ -59,6 +58,28 @@ def get_score_dao(score_id, student_id, exam_seq, page, page_size, db):
     ]
     return total, res_list
 
+def get_all_scores_dao(page, page_size, db):
+    r = db.query(Score).filter(Score.is_deleted == 0)\
+           .order_by(Score.student_id, Score.exam_seq)
+    total = r.count()                                    # 总条数
+    rows = r.offset((page-1)*page_size).limit(page_size).all()# 分页
+    res_list_all = [
+        {
+            "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_all
+
+
 
 def delete_score_dao(score_id, db):
     """逻辑删除，更新is_deleted=1"""

main.py

@@ -26,9 +26,11 @@ app.include_router(example_router)
 app.include_router(statistics_router, prefix="/statistics")
 
 #-------------曾凯---------------
-
+from api.score import scores_router
+app.include_router(scores_router, tags=['成绩接口'])
 #-------------南方宇---------------
-
+from api.employment import employment_router
+app.include_router(employment_router, prefix="/employment")
 #-----------主程序----------------
 
 if __name__ == "__main__":

models/employment.py

@@ -1,13 +1,13 @@
 from sqlalchemy import *
 from sqlalchemy.orm import declarative_base,sessionmaker
 from datetime import datetime
-from core.database import Base, engine
+from core.database import Base
 
 # 为了符合三范式，就业表只保存 student_id，不保存学生姓名和班级名称。查询返回时通过关联 students 和 classes 带出学生姓名、班级名称。
 
 class Employment(Base):
-    __tablename__ = "employment_info_detail"                              # 实际数据库的表名字
-    id = Column(Integer, primary_key=True                   # 主键
+    __tablename__ = "employment_info_detail"                   # 实际数据库的表名字
+    id = Column(Integer, primary_key=True                      # 主键
                 , autoincrement=True                           # 声明是自增主键
                 ,comment="就业信息主键id")
     student_id = Column(Integer, ForeignKey("student_info_detail.id")  # 外键 -> students.id

schemas/employment.py

@@ -1,4 +1,4 @@
-from pydantic import BaseModel, Field
+from pydantic import BaseModel, Field,ConfigDict
 from datetime import date, datetime
 from decimal import Decimal
 
@@ -36,3 +36,4 @@ class EmploymentResponse(BaseModel):
     remark: str | None
     created_at: datetime | None
     updated_at: datetime | None
+    model_config = ConfigDict(from_attributes=True) #允许pydantic从对象属性里取值