iTi-System/iti_system/routes/dict.py

from __future__ import annotations

from fastapi import APIRouter, Depends
from sqlalchemy import select
from sqlalchemy.orm import Session, selectinload

from iti.auth import require_permission
from iti.db import get_db
from iti.responses import ok, page

from iti_system.models import SysDictData, SysDictType
from iti_system.schemas import (
    DictDataCreate,
    DictDataQuery,
    DictDataUpdate,
    DictTypeCreate,
    DictTypeQuery,
    DictTypeUpdate,
)
from iti_system.services import (
    apply_keyword,
    dump_dict_data,
    dump_dict_type,
    get_or_404,
    paginate_query,
)


router = APIRouter(prefix="/sys/dict", tags=["system.dict"])


@router.get("/type/page", dependencies=[Depends(require_permission("system:dict:list"))])
def page_dict_type(query: DictTypeQuery = Depends(), db: Session = Depends(get_db)):
    result = paginate_query(db, _query_types(query), page=query.page, size=query.size)
    return page([dump_dict_type(item, with_data=True) for item in result["items"]], result["page"])


@router.get("/type", dependencies=[Depends(require_permission("system:dict:list"))])
def list_dict_type(query: DictTypeQuery = Depends(), with_data_list: bool = False, db: Session = Depends(get_db)):
    return ok([dump_dict_type(item, with_data=with_data_list) for item in db.scalars(_query_types(query)).all()])


@router.post("/type", dependencies=[Depends(require_permission("system:dict:create"))])
def create_dict_type(payload: DictTypeCreate, db: Session = Depends(get_db)):
    item = SysDictType(**payload.model_dump())
    db.add(item)
    db.commit()
    db.refresh(item)
    return ok(dump_dict_type(item))


@router.put("/type/{id}", dependencies=[Depends(require_permission("system:dict:edit"))])
def update_dict_type(id: str, payload: DictTypeUpdate, db: Session = Depends(get_db)):
    item = get_or_404(db, SysDictType, id, "字典类型不存在")
    for key, value in payload.model_dump(exclude_unset=True).items():
        setattr(item, key, value)
    db.commit()
    db.refresh(item)
    return ok(dump_dict_type(item))


@router.delete("/type/{id}", dependencies=[Depends(require_permission("system:dict:delete"))])
def delete_dict_type(id: str, db: Session = Depends(get_db)):
    item = get_or_404(db, SysDictType, id, "字典类型不存在")
    db.delete(item)
    db.commit()
    return ok()


@router.get("/data/page", dependencies=[Depends(require_permission("system:dict:list"))])
def page_dict_data(query: DictDataQuery = Depends(), db: Session = Depends(get_db)):
    result = paginate_query(db, _query_data(query), page=query.page, size=query.size)
    return page([dump_dict_data(item) for item in result["items"]], result["page"])


@router.get("/data/list", dependencies=[Depends(require_permission("system:dict:list"))])
def list_dict_data(query: DictDataQuery = Depends(), db: Session = Depends(get_db)):
    return ok([dump_dict_data(item) for item in db.scalars(_query_data(query)).all()])


@router.get("/data", dependencies=[Depends(require_permission("system:dict:list"))])
def get_dict_data_by_code(type_code: str, code: str, db: Session = Depends(get_db)):
    item = db.scalar(select(SysDictData).where(SysDictData.type_code == type_code, SysDictData.code == code))
    if item is None:
        return ok(None)
    return ok(dump_dict_data(item))


@router.get("/data/{id}", dependencies=[Depends(require_permission("system:dict:list"))])
def get_dict_data(id: str, db: Session = Depends(get_db)):
    return ok(dump_dict_data(get_or_404(db, SysDictData, id, "字典数据不存在")))


@router.post("/data", dependencies=[Depends(require_permission("system:dict:create"))])
def create_dict_data(payload: DictDataCreate, db: Session = Depends(get_db)):
    item = SysDictData(**payload.model_dump())
    db.add(item)
    db.commit()
    db.refresh(item)
    return ok(dump_dict_data(item))


@router.put("/data/{id}", dependencies=[Depends(require_permission("system:dict:edit"))])
def update_dict_data(id: str, payload: DictDataUpdate, db: Session = Depends(get_db)):
    item = get_or_404(db, SysDictData, id, "字典数据不存在")
    for key, value in payload.model_dump(exclude_unset=True).items():
        setattr(item, key, value)
    db.commit()
    db.refresh(item)
    return ok(dump_dict_data(item))


@router.delete("/data/batch", dependencies=[Depends(require_permission("system:dict:delete"))])
def delete_dict_data_batch(ids: str, db: Session = Depends(get_db)):
    for item_id in [item.strip() for item in ids.split(",") if item.strip()]:
        item = db.get(SysDictData, item_id)
        if item:
            db.delete(item)
    db.commit()
    return ok()


@router.delete("/data/{id}", dependencies=[Depends(require_permission("system:dict:delete"))])
def delete_dict_data(id: str, db: Session = Depends(get_db)):
    item = get_or_404(db, SysDictData, id, "字典数据不存在")
    db.delete(item)
    db.commit()
    return ok()


def _query_types(query: DictTypeQuery):
    stmt = select(SysDictType).options(selectinload(SysDictType.data_list)).order_by(SysDictType.sort, SysDictType.type_name)
    stmt = apply_keyword(stmt, query.keyword, SysDictType.type_name, SysDictType.type_code, SysDictType.desc)
    if query.type_code:
        stmt = stmt.where(SysDictType.type_code.like(f"%{query.type_code}%"))
    if query.status:
        stmt = stmt.where(SysDictType.status == query.status)
    return stmt


def _query_data(query: DictDataQuery):
    stmt = select(SysDictData).order_by(SysDictData.sort, SysDictData.label)
    stmt = apply_keyword(stmt, query.keyword, SysDictData.label, SysDictData.code, SysDictData.value)
    if query.type_code:
        stmt = stmt.where(SysDictData.type_code == query.type_code)
    if query.code:
        stmt = stmt.where(SysDictData.code.like(f"%{query.code}%"))
    if query.status:
        stmt = stmt.where(SysDictData.status == query.status)
    return stmt