Files
FastapiAdmin/backend/app/api/v1/module_system/versions/controller.py
T
zhangtao 0ce31936aa chore: 清理冗余代码与配置,优化项目结构
1.  删除无用文件与废弃代码:移除locale枚举、element-plus插件、sse路由、api token模块等
2.  简化类型导入与依赖:移除大量未使用的类型导入,统一echarts导入方式
3.  优化配置与样式:调整gitignore、样式引入顺序,新增列表动画样式
4.  修复接口与模型:修正接口返回类型、查询参数配置,更新部门模型字段
5.  优化性能与体验:添加图片懒加载,优化加载逻辑与表格渲染
6.  调整环境配置:新增并更新开发/生产环境配置文件
2026-07-23 20:49:52 +08:00

101 lines
4.5 KiB
Python

from typing import Annotated
from fastapi import APIRouter, Body, Depends, Path, Security, status
from fastapi.responses import JSONResponse
from sqlalchemy.ext.asyncio import AsyncSession
from app.common.response import ResponseSchema, SuccessResponse
from app.core.base_schema import AuthSchema, PageResultSchema, PaginationQueryParam
from app.core.dependencies import AuthPermission, db_getter
from app.core.router_class import OperationLogRoute
from .schema import (
VersionCreateSchema,
VersionOutSchema,
VersionQueryParam,
VersionStatusSchema,
VersionUpdateSchema,
)
from .service import VersionService
VersionRouter = APIRouter(route_class=OperationLogRoute, prefix="/versions", tags=["版本管理"])
@VersionRouter.get("/list", summary="分页查询版本", response_model=ResponseSchema[PageResultSchema[VersionOutSchema]])
async def get_version_list_controller(
auth: Annotated[AuthSchema, Security(AuthPermission(["module_system:version:query"]))],
db: Annotated[AsyncSession, Depends(db_getter)],
page: Annotated[PaginationQueryParam, Depends()],
search: Annotated[VersionQueryParam, Depends()],
) -> JSONResponse:
service = VersionService(auth, db)
result = await service.page(page_no=page.page_no, page_size=page.page_size, search=search)
return SuccessResponse(data=result, msg="查询版本列表成功")
@VersionRouter.get("/published", summary="已发布版本列表", response_model=ResponseSchema[list[VersionOutSchema]])
async def get_published_versions_controller(
db: Annotated[AsyncSession, Depends(db_getter)],
) -> JSONResponse:
auth = AuthSchema()
service = VersionService(auth, db)
result = await service.get_published()
return SuccessResponse(data=result, msg="查询成功")
@VersionRouter.get("/detail/{id}", summary="获取版本详情", response_model=ResponseSchema[VersionOutSchema])
async def get_version_detail_controller(
auth: Annotated[AuthSchema, Security(AuthPermission(["module_system:version:detail"]))],
db: Annotated[AsyncSession, Depends(db_getter)],
id: Annotated[int, Path(description="版本ID")],
) -> JSONResponse:
service = VersionService(auth, db)
result = await service.detail(id=id)
return SuccessResponse(data=result, msg="获取版本详情成功")
@VersionRouter.post("/create", status_code=status.HTTP_201_CREATED, summary="创建版本", response_model=ResponseSchema[VersionOutSchema])
async def create_version_controller(
auth: Annotated[AuthSchema, Security(AuthPermission(["module_system:version:create"]))],
db: Annotated[AsyncSession, Depends(db_getter)],
data: Annotated[VersionCreateSchema, Body(description="创建参数")],
) -> JSONResponse:
service = VersionService(auth, db)
result = await service.create(data=data)
return SuccessResponse(data=result, msg="创建版本成功")
@VersionRouter.put("/update/{id}", summary="修改版本", response_model=ResponseSchema[VersionOutSchema])
async def update_version_controller(
auth: Annotated[AuthSchema, Security(AuthPermission(["module_system:version:update"]))],
db: Annotated[AsyncSession, Depends(db_getter)],
id: Annotated[int, Path(description="版本ID")],
data: Annotated[VersionUpdateSchema, Body(description="修改参数")],
) -> JSONResponse:
service = VersionService(auth, db)
result = await service.update(id=id, data=data)
return SuccessResponse(data=result, msg="修改版本成功")
@VersionRouter.delete("/delete", summary="删除版本", response_model=ResponseSchema[None])
async def delete_version_controller(
auth: Annotated[AuthSchema, Security(AuthPermission(["module_system:version:delete"]))],
db: Annotated[AsyncSession, Depends(db_getter)],
ids: Annotated[list[int], Body(description="ID列表")],
) -> JSONResponse:
service = VersionService(auth, db)
await service.delete(ids=ids)
return SuccessResponse(msg="删除版本成功")
@VersionRouter.put("/{id}/status", summary="变更版本状态", response_model=ResponseSchema[VersionOutSchema])
async def set_version_status_controller(
auth: Annotated[AuthSchema, Security(AuthPermission(["module_system:version:update"]))],
db: Annotated[AsyncSession, Depends(db_getter)],
id: Annotated[int, Path(description="版本ID")],
data: Annotated[VersionStatusSchema, Body(description="状态参数")],
) -> JSONResponse:
service = VersionService(auth, db)
result = await service.set_status(id=id, data=data)
return SuccessResponse(data=result, msg="状态变更成功")