test

api/employment.py

@@ -1,13 +1,6 @@
-# APIRouter：路由对象，用来分组接口
 from fastapi import APIRouter, Depends, HTTPException, Query
-# Session：数据库会话类型注解
 from sqlalchemy.orm import Session
-# 类型提示：可选参数
-from typing import Optional
-
-# 导入get_db，数据库会话依赖函数，每次请求生成/释放数据库连接
 from core.database import get_db
-# 导入dao层全部数据库操作函数
 from dao.employment import (
     create_employment,
     get_employment_by_id,
@@ -16,75 +9,54 @@ from dao.employment import (
     update_employment,
     delete_employment_logic
 )
-# 导入pydantic请求体、响应体模型
 from schemas.employment import (
     EmploymentCreate,
     EmploymentUpdate,
-    EmploymentResponse,
+    EmploymentResponse
-    EmploymentListResponse
 )
 
 employment_router = APIRouter(prefix="/employments", tags=["就业管理模块"])
 
 
-@employment_router.post("", response_model=EmploymentResponse, summary="新增就业信息")
+@employment_router.post("/", response_model=EmploymentResponse, summary="新增就业信息")
 def add_employment(body: EmploymentCreate, db: Session = Depends(get_db)):
     """
     新增就业信息
-    - 一个学生只能保存一条就业记录（数据库唯一约束控制）
+    - 一个学生只能保存一条就业记录
-    - student_id必须在学生表student_info_detail中真实存在，否则外键报错
+    - student_id必须在学生表student_info_detail中真实存在
-    :param body:POST请求的JSON请求体，自动解析为EmploymentCreate对象
-    :param db: Depends(get_db)，依赖注入，自动获取数据库会话，请求结束自动关闭连接
-    :return: 返回EmploymentResponse结构数据给前端
     """
-    # 调用dao层新增函数
     res = create_employment(db, body)
-    # dao返回None，代表新增出错（外键不存在/学生重复/数据库异常）
     if res is None:
-        # 抛出HTTP异常，500状态码，返回提示信息给前端/Swagger
         raise HTTPException(status_code=500, detail="新增失败，学生id不存在或该学生已有就业记录")
-    # 成功，直接返回ORM对象，FastAPI会按照response_model自动转为json返回
     return res
 
 
 @employment_router.get("/students/{student_id}", response_model=EmploymentResponse, summary="查询指定学生的就业信息")
 def query_student_employment(student_id: int, db: Session = Depends(get_db)):
-    """
+    """根据学生id获取对应就业记录"""
-    根据学生id查询该学生的就业记录
-    :param student_id: url路径中的路径参数，学生id
-    :param db:数据库会话
-    """
-    # 调用dao，按学生id查询
     record = get_employment_by_student_id(db, student_id)
-    # 如果查询结果为空，抛出404找不到资源
     if not record:
         raise HTTPException(status_code=404, detail="未找到该学生的就业信息")
     return record
 
 
-@employment_router.get("", response_model=EmploymentListResponse, summary="就业信息列表，支持筛选")
+@employment_router.get("/", summary="就业信息列表，支持筛选")
 def query_employment_list(
-    # Query：获取url中?后面的查询参数；description显示在Swagger文档
+    company_name: str | None = Query(None, description="公司名称模糊查询"),
-    company_name: Optional[str] = Query(None, description="公司名称模糊查询"),
+    salary_min: float | None = Query(None, description="最低薪资"),
-    salary_min: Optional[float] = Query(None, description="最低薪资"),
+    salary_max: float | None = Query(None, description="最高薪资"),
-    salary_max: Optional[float] = Query(None, description="最高薪资"),
+    skip: int = Query(0, ge=0, description="偏移量"),
-    skip: int = Query(0, ge=0, description="偏移量"),          # ge=0参数校验，不能传负数
+    limit: int = Query(20, ge=1, le=100, description="每页条数"),
-    limit: int = Query(20, ge=1, le=100, description="每页条数"), # 1‑100条限制，防止一次性查大量数据
     db: Session = Depends(get_db)
 ):
-    """就业分页列表接口，支持公司名模糊，薪资区间筛选"""
+    # 调用dao列表查询，拿到数据字典列表和总条数
-    # 调用dao层列表查询，拿到记录列表和总条数
     data_list, total = get_employment_list(db, company_name, salary_min, salary_max, skip, limit)
-    # 返回符合EmploymentListResponse格式的字典，FastAPI自动序列化
     return {"total": total, "data": data_list}
 
 
 @employment_router.get("/{eid}", response_model=EmploymentResponse, summary="根据id查询就业详情")
 def query_employment_detail(eid: int, db: Session = Depends(get_db)):
-    """
+    """根据就业主键id查询单条就业记录"""
-    根据就业主键id查询单条就业详情
-    :param eid: url路径参数，就业记录主键
-    """
     record = get_employment_by_id(db, eid)
     if not record:
         raise HTTPException(status_code=404, detail="就业记录不存在或已被逻辑删除")
@@ -93,11 +65,7 @@ def query_employment_detail(eid: int, db: Session = Depends(get_db)):
 
 @employment_router.put("/{eid}", response_model=EmploymentResponse, summary="修改就业信息")
 def edit_employment(eid: int, body: EmploymentUpdate, db: Session = Depends(get_db)):
-    """
+    """修改就业记录，只传想要修改的字段即可"""
-    修改就业信息；请求体只传需要修改的字段即可，不需要全部传
-    :param eid:待修改记录id，路径参数
-    :param body:PUT请求的JSON请求体
-    """
     res = update_employment(db, eid, body)
     if res is None:
         raise HTTPException(status_code=404, detail="修改失败，记录不存在")
@@ -106,12 +74,9 @@ def edit_employment(eid: int, body: EmploymentUpdate, db: Session = Depends(get_
 
 @employment_router.delete("/{eid}", summary="逻辑删除就业记录")
 def logic_delete_employment(eid: int, db: Session = Depends(get_db)):
-    """
+    """执行逻辑删除，is_deleted置1，不会真正删除数据库行"""
-    逻辑删除接口：不会物理删除数据库行，只设置is_deleted=1
-    DELETE请求，RESTful风格
-    """
     ok = delete_employment_logic(db, eid)
     if not ok:
         raise HTTPException(status_code=404, detail="删除失败，记录不存在")
-    # 返回简单json提示成功
+    # 删除成功返回提示字典
     return {"code": 200, "msg": "逻辑删除成功"}

dao/employment.py

@@ -1,165 +1,115 @@
-# 导入sqlalchemy会话对象，数据库会话，所有数据库操作都依靠Session
 from sqlalchemy.orm import Session
-# 类型提示：可选对象、列表
-from typing import Optional, List
-# 导入数据库ORM模型
 from models.employment import Employment
-# 导入pydantic入参模型
+from schemas.employment import EmploymentCreate, EmploymentUpdate, EmploymentResponse
-from schemas.employment import EmploymentCreate, EmploymentUpdate
 
 
-def create_employment(db: Session, obj_in: EmploymentCreate) -> Optional[Employment]:
+def create_employment(db, obj_in):
-    """
+    """新增就业信息：一个学生只能存在一条就业记录"""
-    新增就业信息：一个学生只能存在一条就业记录
-    :param db: 数据库会话对象，由FastAPI依赖注入传入
-    :param obj_in: 新增的入参对象 EmploymentCreate
-    :return: 成功返回ORM对象，发生任何异常返回None
-    """
     try:
-        # model_dump()把pydantic对象转字典；**解包字典，赋值给Employment模型字段，创建ORM对象
+        # obj_in.model_dump()把pydantic请求对象转为字典，解包传给Employment生成ORM对象
         db_obj = Employment(**obj_in.model_dump())
-        # 将对象加入会话，此时还没有写入数据库
+        # add：把对象加入数据库会话，此时还没写入数据库
         db.add(db_obj)
-        # 提交事务，真正执行insert SQL写入数据库
+        # commit：提交事务，真正写入mysql数据库
         db.commit()
-        # db.refresh：从数据库刷新对象，回填数据库自动生成的id等字段到db_obj
+        # refresh：从数据库刷新对象，拿到数据库自动生成的id、默认时间等字段
         db.refresh(db_obj)
-        return db_obj
+        # model_validate：把sqlalchemy ORM对象解析进pydantic响应模型
+        # model_dump：把pydantic模型转为普通python字典返回给上层api
+        return EmploymentResponse.model_validate(db_obj).model_dump()
     except Exception:
-        # 只要出现任意异常，执行回滚，撤销本次会话所有未提交操作，防止脏数据残留
+        # 发生任何异常，执行回滚，撤销本次会话中未提交的数据库改动
         db.rollback()
+        # 异常情况返回None，上层api可以判断新增失败
         return None
 
 
-def get_employment_by_id(db: Session, eid: int) -> Optional[Employment]:
+def get_employment_by_id(db, eid):
-    """
+    """根据主键id查询就业详情，过滤逻辑删除"""
-    根据主键id查询就业详情，过滤逻辑删除的数据
-    :param db: 数据库会话
-    :param eid:就业记录主键id
-    :return:查到返回ORM对象；没找到/异常返回None
-    """
     try:
-        # 查询Employment表，条件：id等于eid，并且is_deleted=0（没有被逻辑删除）
+        db_obj = db.query(Employment).filter(
-        return db.query(Employment).filter(
             Employment.id == eid,
             Employment.is_deleted == 0
-        ).first()  # first()获取第一条结果，没有匹配返回None
+        ).first()
+        if not db_obj:
+            return None
+        return EmploymentResponse.model_validate(db_obj).model_dump()
     except Exception:
         return None
 
 
-def get_employment_by_student_id(db: Session, student_id: int) -> Optional[Employment]:
+def get_employment_by_student_id(db, student_id):
-    """
+    """根据学生id查询就业信息，过滤逻辑删除"""
-    根据学生id查询就业信息，过滤逻辑删除
-    :param db:数据库会话
-    :param student_id:学生主键id
-    :return:查到返回ORM对象，否则None
-    """
     try:
-        return db.query(Employment).filter(
+        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()
     except Exception:
         return None
 
 
-def get_employment_list(
+def get_employment_list(db, company_name=None, salary_min=None, salary_max=None, skip=0, limit=20):
-    db: Session,
+    """就业列表查询；支持公司名模糊、薪资范围筛选；只查询is_deleted=0"""
-    company_name: Optional[str] = None,
-    salary_min: Optional[float] = None,
-    salary_max: Optional[float] = None,
-    skip: int = 0,
-    limit: int = 20
-) -> tuple[List[Employment], int]:
-    """
-    就业列表查询；支持公司名模糊、薪资范围筛选；只查询is_deleted=0正常数据
-    :param db:数据库会话
-    :param company_name:公司名称，模糊查询，不传不做过滤
-    :param salary_min:薪资下限，不传不做过滤
-    :param salary_max:薪资上限，不传不做过滤
-    :param skip:分页偏移量，跳过多少条
-    :param limit:分页每页最多取多少条
-    :return:元组 (记录列表,总条数)；异常返回 [],0
-    """
     try:
-        # 构建基础查询对象，只查询未被逻辑删除的数据
         query = db.query(Employment).filter(Employment.is_deleted == 0)
-
-        # 如果传入公司名，增加模糊匹配条件，%代表通配符
         if company_name:
             query = query.filter(Employment.company_name.like(f"%{company_name}%"))
-
-        # 如果传入最低薪资，增加薪资>=下限条件
         if salary_min is not None:
             query = query.filter(Employment.salary >= salary_min)
-
-        # 如果传入最高薪资，增加薪资<=上限条件
         if salary_max is not None:
             query = query.filter(Employment.salary <= salary_max)
-
+        # count()统计满足条件总条数，用于分页total
-        # count()统计符合条件的总记录数，用于前端分页
         total = query.count()
-        # offset跳过skip条，limit最多取limit条，执行查询拿到结果列表
+        # offset偏移量（跳过多少条），limit每页多少条，执行查询拿到结果列表
         records = query.offset(skip).limit(limit).all()
-        return records, total
+        # 列表推导式：循环每一条ORM记录，全部转为字典，得到字典列表
+        dict_list = [EmploymentResponse.model_validate(item).model_dump() for item in records]
+        # 返回字典列表 + 总条数
+        return dict_list, total
     except Exception:
-        # 异常返回空列表、总条数0
+        # 异常返回空列表，总条数0
         return [], 0
 
 
-def update_employment(db: Session, eid: int, obj_in: EmploymentUpdate) -> Optional[Employment]:
+def update_employment(db, eid, obj_in):
-    """
+    """修改就业信息，只更新传入的字段"""
-    修改就业信息，只更新前端传入的字段
-    :param db:数据库会话
-    :param eid:待修改记录主键id
-    :param obj_in:修改入参对象 EmploymentUpdate
-    :return:成功返回更新后的ORM对象；记录不存在/异常返回None
-    """
     try:
-        # 先根据id查询这条记录（自动过滤逻辑删除）
+        # 根据id查询未删除的记录
-        db_obj = get_employment_by_id(db, eid)
+        db_obj_raw = db.query(Employment).filter(
-        # 如果记录不存在，直接返回None
+            Employment.id == eid,
-        if not db_obj:
+            Employment.is_deleted == 0
+        ).first()
+        # 记录不存在直接返回None
+        if db_obj_raw is None:
             return None
-
+        # exclude_unset=True：只取出前端实际传过来的字段，不会带上没传的None字段
-        # exclude_unset=True：只拿到前端实际传过来的字段，没有传的字段不会出现在字典中
         update_dict = obj_in.model_dump(exclude_unset=True)
-
-        # 循环字典，把每一个key‑value赋值给ORM对象属性
         for key, value in update_dict.items():
-            setattr(db_obj, key, value)
+            setattr(db_obj_raw, key, value)
-
-        # 提交事务，把修改写入数据库
         db.commit()
-        # 刷新对象，同步数据库最新数据到对象
+        db.refresh(db_obj_raw)
-        db.refresh(db_obj)
+        return EmploymentResponse.model_validate(db_obj_raw).model_dump()
-        return db_obj
     except Exception:
-        # 发生异常回滚，放弃修改
         db.rollback()
         return None
 
 
-def delete_employment_logic(db: Session, eid: int) -> bool:
+def delete_employment_logic(db, eid):
-    """
+    """逻辑删除，只修改is_deleted=1，不做物理删除"""
-    逻辑删除，只修改is_deleted=1，**不做物理删除db.delete()**，符合项目文档要求
-    :param db:数据库会话
-    :param eid:待删除记录id
-    :return:成功返回True；记录不存在/异常返回False
-    """
     try:
-        # 查询这条记录，过滤已逻辑删除
+        db_obj_raw = db.query(Employment).filter(
-        db_obj = get_employment_by_id(db, eid)
+            Employment.id == eid,
-        # 记录不存在返回False
+            Employment.is_deleted == 0
-        if not db_obj:
+        ).first()
+        if db_obj_raw is None:
             return False
-        # 修改逻辑删除标记为1
+        db_obj_raw.is_deleted = 1
-        db_obj.is_deleted = 1
-        # 提交事务，把修改写入数据库
         db.commit()
         return True
     except Exception:
-        # 异常回滚，返回False
         db.rollback()
         return False

schemas/employment.py

@@ -1,31 +1,27 @@
 from pydantic import BaseModel, Field
-from typing import Optional
+from datetime import date, datetime
-from datetime import date,datetime
 from decimal import Decimal
-# int
 
 
 class EmploymentCreate(BaseModel):
     """新增就业信息 请求体 Schema"""
-    # 学生ID 必填，一个学生只能有一条就业记录
     student_id: int = Field(..., description="关联学生ID，必填")
-    # 就业状态必填，限定可选值业务枚举
+    employment_status: str | None = Field(None, description="就业状态：not_started/job_hunting/offered/employed")
-    employment_status: str = Field(None, description="就业状态：not_started/job_hunting/offered/employed")
+    employment_open_date: date | None = Field(None, description="就业开放时间，选填")
-    employment_open_date: Optional[date] = Field(None, description="就业开放时间，选填")
+    offer_date: date | None = Field(None, description="offer下发时间，选填")
-    offer_date: Optional[date] = Field(None, description="offer下发时间，选填")
+    company_name: str | None = Field(None, max_length=100, description="就业公司名称，选填")
-    company_name: Optional[str] = Field(None, max_length=100, description="就业公司名称，选填")
+    salary: Decimal | None = Field(None, ge=0, description="就业薪资，不能负数，选填")
-    salary: Optional[Decimal] = Field(None, ge=0, description="就业薪资，不能负数，选填")
+    remark: str | None = Field(None, max_length=255, description="备注，选填")
-    remark: Optional[str] = Field(None, max_length=255, description="备注，选填")
 
 
 class EmploymentUpdate(BaseModel):
     """修改就业信息 请求体 Schema：全部字段可选，只传要修改的字段"""
-    employment_status: Optional[str] = Field(None, description="就业状态")
+    employment_status: str | None = Field(None, description="就业状态")
-    employment_open_date: Optional[date] = Field(None, description="就业开放时间")
+    employment_open_date: date | None = Field(None, description="就业开放时间")
-    offer_date: Optional[date] = Field(None, description="offer下发时间")
+    offer_date: date | None = Field(None, description="offer下发时间")
-    company_name: Optional[str] = Field(None, max_length=100, description="就业公司名称")
+    company_name: str | None = Field(None, max_length=100, description="就业公司名称")
-    salary: Optional[Decimal] = Field(None, ge=0, description="就业薪资")
+    salary: Decimal | None = Field(None, ge=0, description="就业薪资")
-    remark: Optional[str] = Field(None, max_length=255, description="备注")
+    remark: str | None = Field(None, max_length=255, description="备注")
 
 
 class EmploymentResponse(BaseModel):
@@ -33,17 +29,10 @@ class EmploymentResponse(BaseModel):
     id: int
     student_id: int
     employment_status: str
-    employment_open_date: Optional[date]
+    employment_open_date: date | None
-    offer_date: Optional[date]
+    offer_date: date | None
-    company_name: Optional[str]
+    company_name: str | None
-    salary: Optional[Decimal]
+    salary: Decimal | None
-    remark: Optional[str]
+    remark: str | None
-    created_at: date
+    created_at: datetime | None
-    updated_at: date
+    updated_at: datetime | None
-
-    class Config:
-        from_attributes = True
-
-class EmploymentListResponse(BaseModel):
-    total: int
-    data: list[EmploymentResponse]