#!/usr/bin/env python # -*- coding: utf-8 -*- """ Dict API - 字典管理接口 提供字典的 CRUD 操作 """ from typing import List, Optional from fastapi import APIRouter, Depends, HTTPException, Query, UploadFile, File from fastapi.responses import StreamingResponse from sqlalchemy import or_, select from sqlalchemy.ext.asyncio import AsyncSession from app.database import get_db from app.config import settings from app.base_schema import PaginatedResponse, ResponseModel from core.application.model import Application from core.dict.model import Dict from core.dict.schema import ( DictCreate, DictUpdate, DictResponse, DictSimple, DictBatchDeleteIn, DictBatchDeleteOut, DictBatchUpdateStatusIn, DictBatchUpdateStatusOut, DictSearchRequest ) from core.dict.service import DictService router = APIRouter(prefix="/dict", tags=["字典管理"]) @router.post("", response_model=DictResponse, summary="创建字典") async def create_dict(data: DictCreate, db: AsyncSession = Depends(get_db)): """创建字典""" # 唯一性校验 if not await DictService.check_unique(db, field="code", value=data.code): raise HTTPException(status_code=400, detail=f"字典编码已存在: {data.code}") dict_obj = await DictService.create(db=db, data=data) return dict_obj @router.get("/all", response_model=List[DictSimple], summary="获取所有字典（简化版）") async def get_all_dicts( application_id: str = Query(None, alias="applicationId", description="所属应用ID"), db: AsyncSession = Depends(get_db), ): """获取所有启用的字典（用于选择器）""" dicts = await DictService.get_all_active(db, application_id=application_id) return dicts @router.get("", response_model=PaginatedResponse[DictResponse], summary="获取字典列表") async def get_dict_list( page: int = Query(default=1, ge=1, description="页码"), page_size: int = Query(default=settings.PAGE_SIZE, ge=1, le=settings.PAGE_MAX_SIZE, alias="pageSize", description="每页数量"), application_id: str = Query(None, alias="applicationId", description="所属应用ID"), name: Optional[str] = Query(default=None, description="字典名称"), code: Optional[str] = Query(default=None, description="字典编码"), status: Optional[bool] = Query(default=None, description="状态"), db: AsyncSession = Depends(get_db) ): """获取字典列表（分页，自动应用数据权限）""" filters = [] # 应用过滤（同时包含全局可见的字典） if application_id: filters.append(or_( Dict.application_id == application_id, Dict.is_global == True # noqa: E712 )) else: filters.append(or_( Dict.application_id.is_(None), Dict.is_global == True # noqa: E712 )) if name: filters.append(Dict.name.ilike(f"%{name}%")) if code: filters.append(Dict.code.ilike(f"%{code}%")) if status is not None: filters.append(Dict.status == status) # 使用带数据权限的列表查询 items, total = await DictService.get_list_with_data_scope(db, page=page, page_size=page_size, filters=filters) # 批量查询应用名称 app_ids = list({item.application_id for item in items if item.application_id}) app_name_map = {} if app_ids: app_result = await db.execute( select(Application.id, Application.name).where(Application.id.in_(app_ids)) ) app_name_map = {row.id: row.name for row in app_result} for item in items: item.application_name = app_name_map.get(item.application_id, "") return PaginatedResponse(items=items, total=total) @router.post("/batch/delete", response_model=DictBatchDeleteOut, summary="批量删除字典") async def batch_delete_dicts( data: DictBatchDeleteIn, db: AsyncSession = Depends(get_db) ): """批量删除字典""" count, failed_ids = await DictService.batch_delete(db, data.ids) return DictBatchDeleteOut(count=count, failed_ids=failed_ids) @router.post("/batch/update_status", response_model=DictBatchUpdateStatusOut, summary="批量更新字典状态") async def batch_update_dict_status( data: DictBatchUpdateStatusIn, db: AsyncSession = Depends(get_db) ): """批量更新字典状态""" count = await DictService.batch_update_status(db, data.ids, data.status) return DictBatchUpdateStatusOut(count=count) @router.post("/search", response_model=PaginatedResponse[DictResponse], summary="搜索字典") async def search_dicts( data: DictSearchRequest, page: int = Query(default=1, ge=1, description="页码"), page_size: int = Query(default=settings.PAGE_SIZE, ge=1, le=settings.PAGE_MAX_SIZE, alias="pageSize", description="每页数量"), db: AsyncSession = Depends(get_db) ): """搜索字典""" items, total = await DictService.search(db, data.keyword, page, page_size) return PaginatedResponse(items=items, total=total) @router.get("/export/excel", summary="导出字典Excel") async def export_dict_excel(db: AsyncSession = Depends(get_db)): """导出字典到Excel""" output = await DictService.export_to_excel(db) return StreamingResponse( output, media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", headers={"Content-Disposition": "attachment; filename=dict_export.xlsx"} ) @router.get("/import/template", summary="下载字典导入模板") async def download_dict_template(): """下载字典导入模板""" output = DictService.get_import_template() return StreamingResponse( output, media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", headers={"Content-Disposition": "attachment; filename=dict_template.xlsx"} ) @router.post("/import/excel", response_model=ResponseModel, summary="导入字典Excel") async def import_dict_excel( file: UploadFile = File(..., description="Excel文件(.xlsx)"), db: AsyncSession = Depends(get_db) ): """从Excel导入字典""" if not file.filename.endswith(".xlsx"): raise HTTPException(status_code=400, detail="只支持.xlsx格式") content = await file.read() success, fail = await DictService.import_from_excel(db, content) return ResponseModel(message=f"成功{success}条，失败{fail}条", data={"success": success, "fail": fail}) @router.get("/check/unique", response_model=ResponseModel, summary="检查字典唯一性") async def check_dict_unique( field: str = Query(..., description="字段名"), value: str = Query(..., description="字段值"), exclude_id: str = Query(default=None, alias="excludeId", description="排除ID"), db: AsyncSession = Depends(get_db) ): """检查字典字段唯一性""" allowed_fields = ["code"] if field not in allowed_fields: raise HTTPException(status_code=400, detail=f"不支持检查字段: {field}") is_unique = await DictService.check_unique(db, field=field, value=value, exclude_id=exclude_id) return ResponseModel(message="可用" if is_unique else "已存在", data={"unique": is_unique}) @router.get("/by/code/{code}", response_model=DictResponse, summary="根据编码获取字典") async def get_dict_by_code(code: str, db: AsyncSession = Depends(get_db)): """根据编码获取字典""" dict_obj = await DictService.get_by_code(db, code) if dict_obj is None: raise HTTPException(status_code=404, detail=f"字典编码不存在: {code}") return dict_obj @router.get("/{dict_id}", response_model=DictResponse, summary="获取字典详情") async def get_dict_by_id(dict_id: str, db: AsyncSession = Depends(get_db)): """获取字典详情""" dict_obj = await DictService.get_by_id(db, dict_id) if dict_obj is None: raise HTTPException(status_code=404, detail="字典不存在") return dict_obj @router.put("/{dict_id}", response_model=DictResponse, summary="更新字典") async def update_dict(dict_id: str, data: DictUpdate, db: AsyncSession = Depends(get_db)): """更新字典""" # 唯一性校验（排除自身） if data.code and not await DictService.check_unique(db, field="code", value=data.code, exclude_id=dict_id): raise HTTPException(status_code=400, detail=f"字典编码已存在: {data.code}") dict_obj = await DictService.update(db, record_id=dict_id, data=data) if dict_obj is None: raise HTTPException(status_code=404, detail="字典不存在") return dict_obj @router.delete("/{dict_id}", response_model=ResponseModel, summary="删除字典") async def delete_dict( dict_id: str, hard: bool = Query(default=False, description="是否物理删除"), db: AsyncSession = Depends(get_db) ): """删除字典""" success = await DictService.delete(db, record_id=dict_id, hard=hard) if not success: raise HTTPException(status_code=404, detail="字典不存在") return ResponseModel(message="删除成功")