南方宇合并测试

api/employment.py

@@ -0,0 +1,59 @@
+from fastapi import APIRouter, Depends, HTTPException, Query    # 导入fastapi组件
+from core.database import get_db            # 导入获取数据库会话的工具函数
+from dao.employment import *                # 导入dao层函数
+from schemas.employment import *            # 导入schema模型
+from typing import List, Optional           # 导入类型提示
+
+employments_router = APIRouter()
+
+@employments_router.post("/employments", response_model=EmploymentResponse, summary="新增就业信息")
+def create_employment(body: EmploymentCreate             # 请求体入参
+                     ,db= Depends(get_db)):              # 注入数据库会话
+    """新增一条学生就业信息，一个学生只能有一条就业记录"""
+    x=body.model_dump()
+    y=create_employment(x,db)
+    if not y:
+        raise HTTPException(status_code=500, detail='服务器繁忙，请稍后添加！')
+    return {"code":200,"detail":"新增信息成功"}
+
+@employments_router.get("/employments", response_model=List[EmploymentResponse], summary="就业列表查询")
+def get_employment_list(company_name: Optional[str] = Query(None, description="公司名称模糊搜索")
+                        ,salary_min: Optional[float] = Query(None, description="薪资下限")
+                        ,salary_max: Optional[float] = Query(None, description="薪资上限")
+                        ,db: Session = Depends(get_db)):
+    """获取就业列表，支持按公司名称、薪资范围过滤"""
+    return get_employment_list(db=db
+                              ,company_name=company_name
+                              ,salary_min=salary_min
+                              ,salary_max=salary_max)
+
+
+@employments_router.get("/employments/students/{student_id}", response_model=EmploymentResponse, summary="查询某个学生的就业信息")
+def get_student_employment(student_id: int
+                          ,db: Session = Depends(get_db)):
+    """根据学生ID获取该学生对应的就业信息"""
+    res = get_employment_by_student_id(db=db, stu_id=student_id)
+    if not res:
+        raise HTTPException(status_code=404, detail="该学生暂无就业信息")
+    return res
+
+
+@employments_router.put("/employments/{emp_id}", response_model=EmploymentResponse, summary="修改就业信息")
+def update_employment(emp_id: int
+                     ,body: EmploymentUpdate
+                     ,db: Session = Depends(get_db)):
+    """根据id修改就业记录信息"""
+    res = update_employment(db=db, emp_id=emp_id, E=body)
+    if not res:
+        raise HTTPException(status_code=404, detail="就业记录不存在或已删除")
+    return res
+
+
+@employments_router.delete("/employments/{emp_id}", summary="逻辑删除就业记录")
+def delete_employment(emp_id: int
+                     ,db: Session = Depends(get_db)):
+    """逻辑删除，设置is_deleted=1，不会物理删除数据库数据，删除后普通接口查询不到"""
+    ok = delete_employment_logic(db=db, emp_id=emp_id)
+    if not ok:
+        raise HTTPException(status_code=404, detail="就业记录不存在或已删除")
+    return {"msg": "逻辑删除成功", "id": emp_id}

dao/employment.py

@@ -0,0 +1,87 @@
+from sqlalchemy.orm import Session           # 导入sqlalchemy会话对象
+from models.employment import Employment     # 导入就业ORM模型
+from schemas.employment import EmploymentCreate, EmploymentUpdate    # 导入pydantic新增、修改schema
+from typing import Optional                  # 导入类型提示
+
+def create_employment(db: Session, E: EmploymentCreate):   #新增就业记录
+    try:
+        db_obj = Employment(**E.model_dump())
+        db.add(db_obj)
+    except:
+        db.rollback()
+        return False
+    else:
+        db.commit()
+        return db_obj
+
+
+def get_employment_by_id(db: Session, emp_id: int):        #根据主键id查询单条就业记录
+    return db.query(Employment).filter(Employment.id == emp_id       # 匹配id
+                                      ,Employment.is_deleted == 0).first()     # 过滤逻辑删除，只查正常数据
+
+
+
+def get_employment_by_student_id(db: Session, stu_id: int):   #根据学生id查询该学生的就业信息
+    return db.query(Employment).filter(Employment.student_id == stu_id
+                                       ,Employment.is_deleted == 0).first()
+
+
+def get_employment_list(db: Session
+                        ,company_name: Optional[str] = None   # 按公司名称模糊查询
+                        ,salary_min: Optional[float] = None   # 薪资下限
+                        ,salary_max: Optional[float] = None): # 薪资上限
+    q = db.query(Employment).filter(Employment.is_deleted == 0)    # 基础查询，只查询未删除数据
+
+
+    if company_name:    # 如果传入公司名称，做模糊匹配
+        q = q.filter(Employment.company_name.like(f"%{company_name}%"))
+
+    if salary_min is not None:    # 如果传入薪资下限
+        q = q.filter(Employment.salary >= salary_min)
+
+    if salary_max is not None:    # 如果传入薪资上限
+        q = q.filter(Employment.salary <= salary_max)
+
+    return q.all()    # 执行查询返回列表
+
+
+
+def update_employment(db: Session, emp_id: int, E: EmploymentUpdate):    #修改就业记录
+    """
+    :param db: 数据库会话
+    :param emp_id: 需要修改的就业记录id
+    :param obj_in: 修改入参 EmploymentUpdate
+    :return: 修改完成的对象 / None（数据不存在）
+    """
+    # 先查询这条记录是否存在且未删除
+    db_obj = get_employment_by_id(db, emp_id)
+    # 判断记录不存在直接返回None
+    if not db_obj:
+        return None
+    # 获取传入的非空字段字典
+    update_data = E.model_dump(exclude_unset=True)
+    # 循环遍历需要更新的字段，赋值给ORM对象
+    for k, v in update_data.items():
+        setattr(db_obj, k, v)
+    db.commit()
+    # 刷新对象获取最新数据
+    db.refresh(db_obj)
+    return db_obj
+
+
+def delete_employment_logic(db: Session, emp_id: int):
+    """
+    逻辑删除就业记录：不物理删除，设置is_deleted=1
+    :param db: 数据库会话
+    :param emp_id: 就业记录主键id
+    :return: True成功；False记录不存在
+    """
+    # 查询目标记录
+    db_obj = get_employment_by_id(db, emp_id)
+    if not db_obj:
+        return False
+    # 设置逻辑删除标记为1
+    db_obj.is_deleted = 1
+    # 提交事务
+    db.commit()
+    return True

models/employment.py

@@ -6,11 +6,11 @@ from core.database import Base, engine
 # 为了符合三范式，就业表只保存 student_id，不保存学生姓名和班级名称。查询返回时通过关联 students 和 classes 带出学生姓名、班级名称。
 
 class Employment(Base):
-    __tablename__ = "employments"                              # 实际数据库的表名字
+    __tablename__ = "employments_info_detail"                              # 实际数据库的表名字
     id = Column(BigInteger, primary_key=True                              # 主键
                 , autoincrement=True                                      # 声明是自增主键
                 , comment="就业信息主键id")
-    student_id = Column(BigInteger, ForeignKey("students.id")  # 外键 -> students.id
+    student_id = Column(BigInteger, ForeignKey("student_info_detail.id")  # 外键
                         , nullable=False                               # 不允许为null
                         , comment="关联学生id")
     employment_status = Column(String(50)

schemas/employment.py

@@ -0,0 +1,43 @@
+from pydantic import BaseModel, Field
+from typing import Optional
+from datetime import date
+from decimal import Decimal
+
+
+class EmploymentCreate(BaseModel):
+    """新增就业信息 请求体 Schema"""
+    # 学生ID 必填，一个学生只能有一条就业记录
+    student_id: int = Field(..., description="关联学生ID，必填")
+    # 就业状态必填，限定可选值业务枚举
+    employment_status: str = Field(..., 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
+
+