76 lines
2.9 KiB
Python
76 lines
2.9 KiB
Python
from fastapi import APIRouter, Depends, HTTPException
|
|
from sqlalchemy.orm import Session
|
|
from typing import List, Optional
|
|
from decimal import Decimal
|
|
|
|
from database import get_db
|
|
from models import Employment, Student, Class
|
|
from schemas import EmploymentCreate, EmploymentUpdate, EmploymentBase
|
|
|
|
router = APIRouter(prefix="/api/employments", tags=["employment"])
|
|
|
|
|
|
@router.get("/", response_model=List[EmploymentBase])
|
|
def list_employments(
|
|
class_id: Optional[int] = None,
|
|
sid: Optional[int] = None,
|
|
db: Session = Depends(get_db),
|
|
skip: int = 0,
|
|
limit: int = 100,
|
|
):
|
|
query = db.query(Employment).filter(Employment.is_deleted == False)
|
|
if class_id is not None:
|
|
query = query.filter(Employment.class_id == class_id)
|
|
if sid is not None:
|
|
query = query.filter(Employment.sid == sid)
|
|
employments = query.offset(skip).limit(limit).all()
|
|
return employments
|
|
|
|
|
|
@router.get("/{employment_id}", response_model=EmploymentBase)
|
|
def get_employment(employment_id: int, db: Session = Depends(get_db)):
|
|
emp = db.query(Employment).filter(Employment.id == employment_id, Employment.is_deleted == False).first()
|
|
if not emp:
|
|
raise HTTPException(status_code=404, detail="就业记录不存在")
|
|
return emp
|
|
|
|
|
|
@router.post("/", response_model=EmploymentBase)
|
|
def create_employment(data: EmploymentCreate, db: Session = Depends(get_db)):
|
|
if not db.query(Student).filter(Student.id == data.sid, Student.is_deleted == False).first():
|
|
raise HTTPException(status_code=400, detail="学生不存在")
|
|
if not db.query(Class).filter(Class.id == data.class_id, Class.is_deleted == False).first():
|
|
raise HTTPException(status_code=400, detail="班级不存在")
|
|
|
|
existing = db.query(Employment).filter(Employment.sid == data.sid, Employment.is_deleted == False).first()
|
|
if existing:
|
|
raise HTTPException(status_code=400, detail="该学生已有就业记录")
|
|
|
|
emp = Employment(**data.model_dump())
|
|
db.add(emp)
|
|
db.commit()
|
|
db.refresh(emp)
|
|
return emp
|
|
|
|
|
|
@router.put("/{employment_id}", response_model=EmploymentBase)
|
|
def update_employment(employment_id: int, data: EmploymentUpdate, db: Session = Depends(get_db)):
|
|
emp = db.query(Employment).filter(Employment.id == employment_id, Employment.is_deleted == False).first()
|
|
if not emp:
|
|
raise HTTPException(status_code=404, detail="就业记录不存在")
|
|
for key, value in data.model_dump(exclude_unset=True).items():
|
|
setattr(emp, key, value)
|
|
db.commit()
|
|
db.refresh(emp)
|
|
return emp
|
|
|
|
|
|
@router.delete("/{employment_id}")
|
|
def delete_employment(employment_id: int, db: Session = Depends(get_db)):
|
|
emp = db.query(Employment).filter(Employment.id == employment_id, Employment.is_deleted == False).first()
|
|
if not emp:
|
|
raise HTTPException(status_code=404, detail="就业记录不存在")
|
|
emp.is_deleted = True
|
|
db.commit()
|
|
return {"message": "删除成功"}
|