230 lines
8.5 KiB
Python
230 lines
8.5 KiB
Python
from typing import Optional
|
||
from fastapi import APIRouter, Depends, HTTPException, Query, UploadFile, File
|
||
from fastapi.responses import StreamingResponse
|
||
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 zq_demo.schema import DemoCreate, DemoUpdate, DemoResponse
|
||
from zq_demo.service import DemoService
|
||
from zq_demo.model import Demo
|
||
|
||
router = APIRouter(prefix="/demos", tags=["Demo管理"])
|
||
|
||
|
||
@router.post("", response_model=DemoResponse, summary="创建Demo")
|
||
async def create_demo(demo: DemoCreate, db: AsyncSession = Depends(get_db)):
|
||
"""
|
||
创建新Demo（自动记录创建人和部门）
|
||
- **title**: 标题
|
||
- **content**: 内容（可选）
|
||
- **status**: 状态（0=草稿，1=发布，2=归档）
|
||
- **priority**: 优先级（0=低，1=中，2=高）
|
||
- **is_active**: 是否激活（默认true）
|
||
"""
|
||
# 检查标题唯一性
|
||
if not await DemoService.check_unique(db, field="title", value=demo.title):
|
||
raise HTTPException(status_code=400, detail="标题已存在")
|
||
|
||
return await DemoService.create(db=db, data=demo)
|
||
|
||
|
||
@router.get("", response_model=PaginatedResponse[DemoResponse], summary="获取Demo列表")
|
||
async def get_demos(
|
||
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="每页数量"),
|
||
title: Optional[str] = Query(None, description="标题（模糊搜索）"),
|
||
status: Optional[int] = Query(None, description="状态筛选"),
|
||
priority: Optional[int] = Query(None, description="优先级筛选"),
|
||
db: AsyncSession = Depends(get_db)
|
||
):
|
||
"""
|
||
获取Demo列表（分页，自动应用数据权限和字段权限）
|
||
|
||
数据权限：根据用户角色自动过滤数据
|
||
- 超级管理员：查看所有数据
|
||
- 普通用户：根据角色配置的数据范围查看（本人/本部门/本部门及下级/全部）
|
||
|
||
字段权限：根据角色配置隐藏/脱敏字段
|
||
- 必填字段（id、title、status）不会被隐藏
|
||
- 可选字段（content、priority等）可以根据配置隐藏
|
||
"""
|
||
# 构建过滤条件
|
||
filters = []
|
||
if title:
|
||
filters.append(Demo.title.ilike(f"%{title}%"))
|
||
if status is not None:
|
||
filters.append(Demo.status == status)
|
||
if priority is not None:
|
||
filters.append(Demo.priority == priority)
|
||
|
||
# 应用行权限（数据权限）- 自动根据用户角色过滤数据
|
||
items, total = await DemoService.get_list_with_data_scope(
|
||
db=db,
|
||
page=page,
|
||
page_size=page_size,
|
||
filters=filters
|
||
)
|
||
|
||
# 转换为响应格式
|
||
response_items = [DemoResponse.model_validate(item) for item in items]
|
||
|
||
# 应用列权限（字段权限）- 自动从上下文获取角色并过滤字段
|
||
response_dicts = [item.model_dump() for item in response_items]
|
||
filtered_items = await DemoService.apply_field_permissions_auto(
|
||
data=response_dicts,
|
||
db=db
|
||
)
|
||
|
||
return PaginatedResponse(items=filtered_items, total=total)
|
||
|
||
|
||
@router.get("/export/excel", summary="导出Excel")
|
||
async def export_excel(db: AsyncSession = Depends(get_db)):
|
||
"""
|
||
导出Demo数据到Excel（自动应用数据权限）
|
||
|
||
只导出当前用户有权限查看的数据
|
||
"""
|
||
output = await DemoService.export_to_excel_with_data_scope(
|
||
db=db,
|
||
data_converter=DemoService._export_converter
|
||
)
|
||
return StreamingResponse(
|
||
output,
|
||
media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
|
||
headers={"Content-Disposition": "attachment; filename=demo_export.xlsx"}
|
||
)
|
||
|
||
|
||
@router.get("/import/template", summary="下载导入模板")
|
||
async def download_template():
|
||
"""
|
||
下载Excel导入模板
|
||
"""
|
||
output = DemoService.get_import_template()
|
||
return StreamingResponse(
|
||
output,
|
||
media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
|
||
headers={"Content-Disposition": "attachment; filename=demo_template.xlsx"}
|
||
)
|
||
|
||
|
||
@router.post("/import/excel", response_model=ResponseModel, summary="导入Excel")
|
||
async def import_excel(
|
||
file: UploadFile = File(..., description="Excel文件(.xlsx)"),
|
||
db: AsyncSession = Depends(get_db)
|
||
):
|
||
"""
|
||
从Excel导入Demo数据
|
||
"""
|
||
if not file.filename.endswith(".xlsx"):
|
||
raise HTTPException(status_code=400, detail="只支持.xlsx格式的Excel文件")
|
||
|
||
content = await file.read()
|
||
success_count, fail_count = await DemoService.import_from_excel(db, content)
|
||
|
||
return ResponseModel(
|
||
message=f"导入完成，成功{success_count}条，失败{fail_count}条",
|
||
data={"success": success_count, "fail": fail_count}
|
||
)
|
||
|
||
|
||
@router.get("/check/unique", response_model=ResponseModel, summary="检查字段唯一性")
|
||
async def check_unique(
|
||
field: str = Query(..., description="字段名，如title"),
|
||
value: str = Query(..., description="字段值"),
|
||
exclude_id: str = Query(default=None, alias="excludeId", description="排除的记录ID（更新时使用）"),
|
||
db: AsyncSession = Depends(get_db)
|
||
):
|
||
"""
|
||
检查字段值是否唯一
|
||
- **field**: 字段名（如 title）
|
||
- **value**: 要检查的值
|
||
- **excludeId**: 更新时排除自身的ID
|
||
|
||
返回: {"unique": true/false}
|
||
"""
|
||
# 限制可检查的字段，防止恶意查询
|
||
allowed_fields = ["title"]
|
||
if field not in allowed_fields:
|
||
raise HTTPException(status_code=400, detail=f"不支持检查字段: {field}，允许的字段: {allowed_fields}")
|
||
|
||
is_unique = await DemoService.check_unique(db, field=field, value=value, exclude_id=exclude_id)
|
||
return ResponseModel(
|
||
message="字段值可用" if is_unique else "字段值已存在",
|
||
data={"unique": is_unique}
|
||
)
|
||
|
||
|
||
@router.get("/{demo_id}", response_model=DemoResponse, summary="获取单个Demo")
|
||
async def get_demo(demo_id: str, db: AsyncSession = Depends(get_db)):
|
||
"""
|
||
根据Demo ID获取Demo详情（应用字段权限）
|
||
|
||
字段权限：根据角色配置隐藏/脱敏字段
|
||
"""
|
||
db_demo = await DemoService.get_by_id(db, record_id=demo_id)
|
||
if db_demo is None:
|
||
raise HTTPException(status_code=404, detail="Demo不存在")
|
||
|
||
# 转换为响应格式
|
||
response = DemoResponse.model_validate(db_demo)
|
||
|
||
# 应用列权限（字段权限）
|
||
response_dict = response.model_dump()
|
||
filtered_data = await DemoService.apply_field_permissions_auto(
|
||
data=response_dict,
|
||
db=db
|
||
)
|
||
|
||
return filtered_data
|
||
|
||
|
||
@router.put("/{demo_id}", response_model=DemoResponse, summary="全量更新Demo")
|
||
async def update_demo(demo_id: str, demo: DemoUpdate, db: AsyncSession = Depends(get_db)):
|
||
"""
|
||
全量更新Demo信息
|
||
"""
|
||
# 检查标题唯一性（排除自身）
|
||
if demo.title and not await DemoService.check_unique(db, field="title", value=demo.title, exclude_id=demo_id):
|
||
raise HTTPException(status_code=400, detail="标题已存在")
|
||
|
||
db_demo = await DemoService.update(db, record_id=demo_id, data=demo)
|
||
if db_demo is None:
|
||
raise HTTPException(status_code=404, detail="Demo不存在")
|
||
return db_demo
|
||
|
||
|
||
@router.patch("/{demo_id}", response_model=DemoResponse, summary="部分更新Demo")
|
||
async def patch_demo(demo_id: str, demo: DemoUpdate, db: AsyncSession = Depends(get_db)):
|
||
"""
|
||
部分更新Demo信息（只更新传入的字段）
|
||
"""
|
||
# 检查标题唯一性（排除自身）
|
||
if demo.title and not await DemoService.check_unique(db, field="title", value=demo.title, exclude_id=demo_id):
|
||
raise HTTPException(status_code=400, detail="标题已存在")
|
||
|
||
db_demo = await DemoService.update(db, record_id=demo_id, data=demo)
|
||
if db_demo is None:
|
||
raise HTTPException(status_code=404, detail="Demo不存在")
|
||
return db_demo
|
||
|
||
|
||
@router.delete("/{demo_id}", response_model=ResponseModel, summary="删除Demo")
|
||
async def delete_demo(
|
||
demo_id: str,
|
||
hard: bool = Query(default=False, description="是否物理删除，False为逻辑删除，True为物理删除"),
|
||
db: AsyncSession = Depends(get_db)
|
||
):
|
||
"""
|
||
删除Demo
|
||
- **hard=false**: 逻辑删除（默认）
|
||
- **hard=true**: 物理删除
|
||
"""
|
||
success = await DemoService.delete(db, record_id=demo_id, hard=hard)
|
||
if not success:
|
||
raise HTTPException(status_code=404, detail="Demo不存在")
|
||
return ResponseModel(message="删除成功")
|