test

api/employment.py

@@ -0,0 +1,117 @@
+# 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,
+    get_employment_by_student_id,
+    get_employment_list,
+    update_employment,
+    delete_employment_logic
+)
+# 导入pydantic请求体、响应体模型
+from schemas.employment import (
+    EmploymentCreate,
+    EmploymentUpdate,
+    EmploymentResponse,
+    EmploymentListResponse
+)
+
+employment_router = APIRouter(prefix="/employments", tags=["就业管理模块"])
+
+
+@employment_router.post("", response_model=EmploymentResponse, summary="新增就业信息")
+def add_employment(body: EmploymentCreate, db: Session = Depends(get_db)):
+    """
+    新增就业信息
+    - 一个学生只能保存一条就业记录（数据库唯一约束控制）
+    - 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查询该学生的就业记录
+    :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="就业信息列表，支持筛选")
+def query_employment_list(
+    # Query：获取url中?后面的查询参数；description显示在Swagger文档
+    company_name: Optional[str] = Query(None, description="公司名称模糊查询"),
+    salary_min: Optional[float] = Query(None, description="最低薪资"),
+    salary_max: Optional[float] = Query(None, description="最高薪资"),
+    skip: int = Query(0, ge=0, description="偏移量"),          # ge=0参数校验，不能传负数
+    limit: int = Query(20, ge=1, le=100, description="每页条数"), # 1‑100条限制，防止一次性查大量数据
+    db: Session = Depends(get_db)
+):
+    """就业分页列表接口，支持公司名模糊，薪资区间筛选"""
+    # 调用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查询单条就业详情
+    :param eid: url路径参数，就业记录主键
+    """
+    record = get_employment_by_id(db, eid)
+    if not record:
+        raise HTTPException(status_code=404, detail="就业记录不存在或已被逻辑删除")
+    return record
+
+
+@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="修改失败，记录不存在")
+    return res
+
+
+@employment_router.delete("/{eid}", summary="逻辑删除就业记录")
+def logic_delete_employment(eid: int, db: Session = Depends(get_db)):
+    """
+    逻辑删除接口：不会物理删除数据库行，只设置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

@@ -0,0 +1,165 @@
+# 导入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
+
+
+def create_employment(db: Session, obj_in: EmploymentCreate) -> Optional[Employment]:
+    """
+    新增就业信息：一个学生只能存在一条就业记录
+    :param db: 数据库会话对象，由FastAPI依赖注入传入
+    :param obj_in: 新增的入参对象 EmploymentCreate
+    :return: 成功返回ORM对象，发生任何异常返回None
+    """
+    try:
+        # model_dump()把pydantic对象转字典；**解包字典，赋值给Employment模型字段，创建ORM对象
+        db_obj = Employment(**obj_in.model_dump())
+        # 将对象加入会话，此时还没有写入数据库
+        db.add(db_obj)
+        # 提交事务，真正执行insert SQL写入数据库
+        db.commit()
+        # db.refresh：从数据库刷新对象，回填数据库自动生成的id等字段到db_obj
+        db.refresh(db_obj)
+        return db_obj
+    except Exception:
+        # 只要出现任意异常，执行回滚，撤销本次会话所有未提交操作，防止脏数据残留
+        db.rollback()
+        return None
+
+
+def get_employment_by_id(db: Session, eid: int) -> Optional[Employment]:
+    """
+    根据主键id查询就业详情，过滤逻辑删除的数据
+    :param db: 数据库会话
+    :param eid:就业记录主键id
+    :return:查到返回ORM对象；没找到/异常返回None
+    """
+    try:
+        # 查询Employment表，条件：id等于eid，并且is_deleted=0（没有被逻辑删除）
+        return db.query(Employment).filter(
+            Employment.id == eid,
+            Employment.is_deleted == 0
+        ).first()  # first()获取第一条结果，没有匹配返回None
+    except Exception:
+        return None
+
+
+def get_employment_by_student_id(db: Session, student_id: int) -> Optional[Employment]:
+    """
+    根据学生id查询就业信息，过滤逻辑删除
+    :param db:数据库会话
+    :param student_id:学生主键id
+    :return:查到返回ORM对象，否则None
+    """
+    try:
+        return db.query(Employment).filter(
+            Employment.student_id == student_id,
+            Employment.is_deleted == 0
+        ).first()
+    except Exception:
+        return None
+
+
+def get_employment_list(
+    db: Session,
+    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 = query.count()
+        # offset跳过skip条，limit最多取limit条，执行查询拿到结果列表
+        records = query.offset(skip).limit(limit).all()
+        return records, total
+    except Exception:
+        # 异常返回空列表、总条数0
+        return [], 0
+
+
+def update_employment(db: Session, eid: int, obj_in: EmploymentUpdate) -> Optional[Employment]:
+    """
+    修改就业信息，只更新前端传入的字段
+    :param db:数据库会话
+    :param eid:待修改记录主键id
+    :param obj_in:修改入参对象 EmploymentUpdate
+    :return:成功返回更新后的ORM对象；记录不存在/异常返回None
+    """
+    try:
+        # 先根据id查询这条记录（自动过滤逻辑删除）
+        db_obj = get_employment_by_id(db, eid)
+        # 如果记录不存在，直接返回None
+        if not db_obj:
+            return 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)
+
+        # 提交事务，把修改写入数据库
+        db.commit()
+        # 刷新对象，同步数据库最新数据到对象
+        db.refresh(db_obj)
+        return db_obj
+    except Exception:
+        # 发生异常回滚，放弃修改
+        db.rollback()
+        return None
+
+
+def delete_employment_logic(db: Session, eid: int) -> bool:
+    """
+    逻辑删除，只修改is_deleted=1，**不做物理删除db.delete()**，符合项目文档要求
+    :param db:数据库会话
+    :param eid:待删除记录id
+    :return:成功返回True；记录不存在/异常返回False
+    """
+    try:
+        # 查询这条记录，过滤已逻辑删除
+        db_obj = get_employment_by_id(db, eid)
+        # 记录不存在返回False
+        if not db_obj:
+            return False
+        # 修改逻辑删除标记为1
+        db_obj.is_deleted = 1
+        # 提交事务，把修改写入数据库
+        db.commit()
+        return True
+    except Exception:
+        # 异常回滚，返回False
+        db.rollback()
+        return False

schemas/employment.py

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