114 lines
3.7 KiB
Python
114 lines
3.7 KiB
Python
from fastapi import APIRouter, Depends, HTTPException, status
|
|
from sqlalchemy.orm import Session
|
|
from app.core.db import get_db
|
|
from app.schemas.dict import (
|
|
DictTypeOut,
|
|
DictTypeCreate,
|
|
DictTypeUpdate,
|
|
DictItemOut,
|
|
DictItemCreate,
|
|
DictItemUpdate,
|
|
)
|
|
from app.models import DictType, DictItem
|
|
|
|
router = APIRouter(prefix="/dicts", tags=["dicts"])
|
|
|
|
|
|
@router.get("/types", response_model=list[DictTypeOut])
|
|
def list_types(db: Session = Depends(get_db)):
|
|
return db.query(DictType).all()
|
|
|
|
|
|
@router.post("/types", response_model=DictTypeOut)
|
|
def create_type(payload: DictTypeCreate, db: Session = Depends(get_db)):
|
|
exists = db.query(DictType).filter(DictType.type_code == payload.type_code).first()
|
|
if exists:
|
|
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Type code exists")
|
|
item = DictType(
|
|
type_code=payload.type_code,
|
|
type_name=payload.type_name,
|
|
status=payload.status,
|
|
remark=payload.remark,
|
|
)
|
|
db.add(item)
|
|
db.commit()
|
|
db.refresh(item)
|
|
return item
|
|
|
|
|
|
@router.put("/types/{type_id}", response_model=DictTypeOut)
|
|
def update_type(type_id: int, payload: DictTypeUpdate, db: Session = Depends(get_db)):
|
|
item = db.query(DictType).filter(DictType.dict_type_id == type_id).first()
|
|
if not item:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Type not found")
|
|
if payload.type_name is not None:
|
|
item.type_name = payload.type_name
|
|
if payload.status is not None:
|
|
item.status = payload.status
|
|
if payload.remark is not None:
|
|
item.remark = payload.remark
|
|
db.commit()
|
|
db.refresh(item)
|
|
return item
|
|
|
|
|
|
@router.delete("/types/{type_id}")
|
|
def delete_type(type_id: int, db: Session = Depends(get_db)):
|
|
item = db.query(DictType).filter(DictType.dict_type_id == type_id).first()
|
|
if not item:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Type not found")
|
|
db.delete(item)
|
|
db.commit()
|
|
return {"status": "ok"}
|
|
|
|
|
|
@router.get("/items", response_model=list[DictItemOut])
|
|
def list_items(type_code: str, db: Session = Depends(get_db)):
|
|
return db.query(DictItem).filter(DictItem.type_code == type_code).order_by(DictItem.sort_order).all()
|
|
|
|
|
|
@router.post("/items", response_model=DictItemOut)
|
|
def create_item(payload: DictItemCreate, db: Session = Depends(get_db)):
|
|
item = DictItem(
|
|
type_code=payload.type_code,
|
|
item_label=payload.item_label,
|
|
item_value=payload.item_value,
|
|
sort_order=payload.sort_order,
|
|
status=payload.status,
|
|
remark=payload.remark,
|
|
)
|
|
db.add(item)
|
|
db.commit()
|
|
db.refresh(item)
|
|
return item
|
|
|
|
|
|
@router.put("/items/{item_id}", response_model=DictItemOut)
|
|
def update_item(item_id: int, payload: DictItemUpdate, db: Session = Depends(get_db)):
|
|
item = db.query(DictItem).filter(DictItem.dict_item_id == item_id).first()
|
|
if not item:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Item not found")
|
|
if payload.item_label is not None:
|
|
item.item_label = payload.item_label
|
|
if payload.item_value is not None:
|
|
item.item_value = payload.item_value
|
|
if payload.sort_order is not None:
|
|
item.sort_order = payload.sort_order
|
|
if payload.status is not None:
|
|
item.status = payload.status
|
|
if payload.remark is not None:
|
|
item.remark = payload.remark
|
|
db.commit()
|
|
db.refresh(item)
|
|
return item
|
|
|
|
|
|
@router.delete("/items/{item_id}")
|
|
def delete_item(item_id: int, db: Session = Depends(get_db)):
|
|
item = db.query(DictItem).filter(DictItem.dict_item_id == item_id).first()
|
|
if not item:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Item not found")
|
|
db.delete(item)
|
|
db.commit()
|
|
return {"status": "ok"}
|