mirror of
https://github.com/fastapiadmin/FastapiAdmin.git
synced 2026-09-21 12:52:26 +00:00
refactor(generator): 优化代码生成模块接口和服务实现
- 修改接口定义，增加路径参数并完善请求描述，增强参数校验和依赖注入 - 优化CRUD层数据库操作，统一异步会话使用，删除多余db参数 - 增加业务表与字段模型关系级联删除配置，优化模型关联关系声明 - 精简pydantic模型，去除冗余校验装饰器，完善字段描述和必填约束 - 服务层增加类型检查和异常抛出，规范业务逻辑流程和错误提示 - 优化代码结构，调整模块导入顺序和注释，提升代码可读性和一致性
This commit is contained in:
zhangtao
@@ -33,7 +33,7 @@ class DeptCRUD(CRUDBase[DeptModel, DeptCreateSchema, DeptUpdateSchema]):
|
||||
obj.parent_name = parent.name
|
||||
return obj
|
||||
|
||||
async def get_list_crud(self, search: Dict = None, order_by: List[Dict[str, str]] = None) -> Sequence[DeptModel]:
|
||||
async def get_list_crud(self, search: Optional[Dict] = None, order_by: Optional[List[Dict[str, str]]] = None) -> Sequence[DeptModel]:
|
||||
"""
|
||||
获取部门列表
|
||||
|
||||
@@ -51,7 +51,7 @@ class DeptCRUD(CRUDBase[DeptModel, DeptCreateSchema, DeptUpdateSchema]):
|
||||
obj.parent_name = parent_map.get(obj.parent_id)
|
||||
return obj_list
|
||||
|
||||
async def get_tree_list_crud(self, search: Dict = None, order_by: List[Dict[str, str]] = None) -> Sequence[DeptModel]:
|
||||
async def get_tree_list_crud(self, search: Optional[Dict] = None, order_by: Optional[List[Dict[str, str]]] = None) -> Sequence[DeptModel]:
|
||||
"""
|
||||
获取部门树形列表
|
||||
|
||||
@@ -59,7 +59,7 @@ class DeptCRUD(CRUDBase[DeptModel, DeptCreateSchema, DeptUpdateSchema]):
|
||||
:param order_by: 排序字段
|
||||
:return: 部门树形列表
|
||||
"""
|
||||
return await self.get_tree_list(search=search, order_by=order_by, children_attr='children')
|
||||
return await self.tree_list(search=search, order_by=order_by, children_attr='children')
|
||||
|
||||
async def set_available_crud(self, ids: List[int], status: bool) -> None:
|
||||
"""
|
||||
|
||||
@@ -20,7 +20,7 @@ class DictTypeCRUD(CRUDBase[DictTypeModel, DictTypeCreateSchema, DictTypeUpdateS
|
||||
"""获取数据字典类型详情"""
|
||||
return await self.get(id=id)
|
||||
|
||||
async def get_obj_list_crud(self, search: Dict = None, order_by: List[Dict[str, str]] = None) -> Sequence[DictTypeModel]:
|
||||
async def get_obj_list_crud(self, search: Optional[Dict] = None, order_by: Optional[List[Dict[str, str]]] = None) -> Sequence[DictTypeModel]:
|
||||
"""获取数据字典类型列表"""
|
||||
return await self.list(search=search, order_by=order_by)
|
||||
|
||||
@@ -53,7 +53,7 @@ class DictDataCRUD(CRUDBase[DictDataModel, DictDataCreateSchema, DictDataUpdateS
|
||||
"""获取数据字典数据详情"""
|
||||
return await self.get(id=id)
|
||||
|
||||
async def get_obj_list_crud(self, search: Dict = None, order_by: List[Dict[str, str]] = None) -> Sequence[DictDataModel]:
|
||||
async def get_obj_list_crud(self, search: Optional[Dict] = None, order_by: Optional[List[Dict[str, str]]] = None) -> Sequence[DictDataModel]:
|
||||
"""获取数据字典数据列表"""
|
||||
return await self.list(search=search, order_by=order_by)
|
||||
|
||||
|
||||
@@ -34,7 +34,7 @@ class OperationLogCRUD(CRUDBase[OperationLogModel, OperationLogCreateSchema, Non
|
||||
"""
|
||||
return await self.get(id=id)
|
||||
|
||||
async def get_list_crud(self, search: Dict = None, order_by: List[Dict[str, str]] = None) -> Sequence[OperationLogModel]:
|
||||
async def get_list_crud(self, search: Optional[Dict] = None, order_by: Optional[List[Dict[str, str]]] = None) -> Sequence[OperationLogModel]:
|
||||
"""
|
||||
获取操作日志列表
|
||||
|
||||
|
||||
@@ -33,7 +33,7 @@ class MenuCRUD(CRUDBase[MenuModel, MenuCreateSchema, MenuUpdateSchema]):
|
||||
obj.parent_name = parent.name
|
||||
return obj
|
||||
|
||||
async def get_list_crud(self, search: Dict = None, order_by: List[Dict[str, str]] = None) -> Sequence[MenuModel]:
|
||||
async def get_list_crud(self, search: Optional[Dict] = None, order_by: Optional[List[Dict[str, str]]] = None) -> Sequence[MenuModel]:
|
||||
"""
|
||||
获取菜单列表
|
||||
|
||||
@@ -51,7 +51,7 @@ class MenuCRUD(CRUDBase[MenuModel, MenuCreateSchema, MenuUpdateSchema]):
|
||||
obj.parent_name = parent_map.get(obj.parent_id)
|
||||
return obj_list
|
||||
|
||||
async def get_tree_list_crud(self, search: Dict = None, order_by: List[Dict[str, str]] = None) -> Sequence[MenuModel]:
|
||||
async def get_tree_list_crud(self, search: Optional[Dict] = None, order_by: Optional[List[Dict[str, str]]] = None) -> Sequence[MenuModel]:
|
||||
"""
|
||||
获取菜单树形列表
|
||||
|
||||
@@ -59,7 +59,7 @@ class MenuCRUD(CRUDBase[MenuModel, MenuCreateSchema, MenuUpdateSchema]):
|
||||
:param order_by: 排序字段
|
||||
:return: 菜单树形列表
|
||||
"""
|
||||
return await self.get_tree_list(search=search, order_by=order_by, children_attr='children')
|
||||
return await self.tree_list(search=search, order_by=order_by, children_attr='children')
|
||||
|
||||
async def set_available_crud(self, ids: List[int], status: bool) -> None:
|
||||
"""
|
||||
|
||||
@@ -39,7 +39,7 @@ class MenuModel(ModelMixin):
|
||||
keep_alive: Mapped[bool] = mapped_column(Boolean, default=True, nullable=False, comment='是否缓存(True:是 False:否)')
|
||||
always_show: Mapped[bool] = mapped_column(Boolean, default=False, nullable=False, comment='是否始终显示(True:是 False:否)')
|
||||
title: Mapped[Optional[str]] = mapped_column(String(50), comment='菜单标题')
|
||||
params: Mapped[Optional[dict]] = mapped_column(JSON, comment='路由参数(JSON对象)')
|
||||
params: Mapped[Optional[list[dict[str, str]]]] = mapped_column(JSON, comment='路由参数(JSON对象)')
|
||||
affix: Mapped[bool] = mapped_column(Boolean, default=False, nullable=False, comment='是否固定标签页(True:是 False:否)')
|
||||
|
||||
parent_id: Mapped[Optional[int]] = mapped_column(Integer, ForeignKey('system_menu.id', ondelete='SET NULL'), default=None, index=True, comment='父菜单ID')
|
||||
|
||||
@@ -20,7 +20,7 @@ class NoticeCRUD(CRUDBase[NoticeModel, NoticeCreateSchema, NoticeUpdateSchema]):
|
||||
"""获取公告详情"""
|
||||
return await self.get(id=id)
|
||||
|
||||
async def get_list_crud(self, search: Dict = None, order_by: List[Dict[str, str]] = None) -> Sequence[NoticeModel]:
|
||||
async def get_list_crud(self, search: Optional[Dict] = None, order_by: Optional[List[Dict[str, str]]] = None) -> Sequence[NoticeModel]:
|
||||
"""获取公告列表"""
|
||||
return await self.list(search=search, order_by=order_by)
|
||||
|
||||
|
||||
@@ -24,7 +24,7 @@ class ParamsCRUD(CRUDBase[ParamsModel, ParamsCreateSchema, ParamsUpdateSchema]):
|
||||
"""根据key获取配置管理型详情"""
|
||||
return await self.get(config_key=key)
|
||||
|
||||
async def get_obj_list_crud(self, search: Dict = None, order_by: List[Dict[str, str]] = None) -> Sequence[ParamsModel]:
|
||||
async def get_obj_list_crud(self, search: Optional[Dict] = None, order_by: Optional[List[Dict[str, str]]] = None) -> Sequence[ParamsModel]:
|
||||
"""获取配置管理型列表"""
|
||||
return await self.list(search=search, order_by=order_by)
|
||||
|
||||
|
||||
@@ -1,7 +1,7 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
|
||||
import json
|
||||
from typing import Any, Dict, List
|
||||
from typing import Any, Dict, List, Optional
|
||||
|
||||
from redis.asyncio.client import Redis
|
||||
from fastapi import UploadFile
|
||||
@@ -39,7 +39,7 @@ class ParamsService:
|
||||
return ParamsOutSchema.model_validate(obj).model_dump()
|
||||
|
||||
@classmethod
|
||||
async def get_config_value_by_key_service(cls, auth: AuthSchema, config_key: str) -> str:
|
||||
async def get_config_value_by_key_service(cls, auth: AuthSchema, config_key: str) -> str | None:
|
||||
"""根据配置键获取配置值"""
|
||||
obj = await ParamsCRUD(auth).get_obj_by_key_crud(key=config_key)
|
||||
if not obj:
|
||||
@@ -47,9 +47,7 @@ class ParamsService:
|
||||
return obj.config_value
|
||||
|
||||
@classmethod
|
||||
async def get_obj_list_service(cls, auth: AuthSchema, search: ParamsQueryParam = None, order_by: List[Dict[str, str]] = None) -> List[Dict]:
|
||||
if order_by:
|
||||
order_by = eval(order_by)
|
||||
async def get_obj_list_service(cls, auth: AuthSchema, search: Optional[ParamsQueryParam] = None, order_by: Optional[List[Dict[str, str]]]= None) -> List[Dict]:
|
||||
obj_list = None
|
||||
if search:
|
||||
obj_list = await ParamsCRUD(auth).get_obj_list_crud(search=search.__dict__, order_by=order_by)
|
||||
@@ -91,6 +89,8 @@ class ParamsService:
|
||||
raise CustomException(msg='更新失败，系统配置key不允许修改')
|
||||
|
||||
new_obj = await ParamsCRUD(auth).update_obj_crud(id=id, data=data)
|
||||
if not new_obj:
|
||||
raise CustomException(msg='更新失败，系统配置不存在')
|
||||
new_obj_dict = ParamsOutSchema.model_validate(new_obj).model_dump()
|
||||
|
||||
# 同步redis
|
||||
@@ -171,7 +171,7 @@ class ParamsService:
|
||||
).model_dump()
|
||||
|
||||
@classmethod
|
||||
async def init_config_service(cls, redis: Redis) -> bool:
|
||||
async def init_config_service(cls, redis: Redis) -> None:
|
||||
async with AsyncSessionLocal() as session:
|
||||
async with session.begin():
|
||||
auth = AuthSchema(db=session)
|
||||
|
||||
@@ -29,7 +29,10 @@ async def get_obj_list_controller(
|
||||
search: PositionQueryParam = Depends(),
|
||||
auth: AuthSchema = Depends(AuthPermission(permissions=["system:position:query"])),
|
||||
) -> JSONResponse:
|
||||
result_dict_list = await PositionService.get_position_list_service(search=search, auth=auth, order_by=page.order_by)
|
||||
order_by = [{"order": "asc"}]
|
||||
if page.order_by:
|
||||
order_by = page.order_by
|
||||
result_dict_list = await PositionService.get_position_list_service(search=search, auth=auth, order_by=order_by)
|
||||
result_dict = await PaginationService.paginate(data_list= result_dict_list, page_no= page.page_no, page_size = page.page_size)
|
||||
logger.info(f"查询岗位列表成功")
|
||||
return SuccessResponse(data=result_dict, msg="查询岗位列表成功")
|
||||
|
||||
@@ -25,7 +25,7 @@ class PositionCRUD(CRUDBase[PositionModel, PositionCreateSchema, PositionUpdateS
|
||||
"""
|
||||
return await self.get(id=id)
|
||||
|
||||
async def get_list_crud(self, search: Dict = None, order_by: List[Dict[str, str]] = None) -> Sequence[PositionModel]:
|
||||
async def get_list_crud(self, search: Optional[Dict] = None, order_by: Optional[List[Dict[str, str]]] = None) -> Sequence[PositionModel]:
|
||||
"""
|
||||
获取岗位列表
|
||||
|
||||
|
||||
@@ -28,6 +28,6 @@ class PositionQueryParam:
|
||||
|
||||
# 时间范围查询
|
||||
if start_time and end_time:
|
||||
start_datetime = datetime.strptime(start_time, '%Y-%m-%d %H:%M:%S')
|
||||
end_datetime = datetime.strptime(end_time, '%Y-%m-%d %H:%M:%S')
|
||||
start_datetime = datetime.strptime(str(start_time), '%Y-%m-%d %H:%M:%S')
|
||||
end_datetime = datetime.strptime(str(end_time), '%Y-%m-%d %H:%M:%S')
|
||||
self.created_at = ("between", (start_datetime, end_datetime))
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
|
||||
from typing import Any, Dict, List
|
||||
from typing import Any, Dict, List, Optional
|
||||
|
||||
from app.core.base_schema import BatchSetAvailable
|
||||
from app.core.exceptions import CustomException
|
||||
@@ -25,12 +25,8 @@ class PositionService:
|
||||
return PositionOutSchema.model_validate(position).model_dump()
|
||||
|
||||
@classmethod
|
||||
async def get_position_list_service(cls, auth: AuthSchema, search: PositionQueryParam, order_by: List[Dict] = None) -> List[Dict]:
|
||||
async def get_position_list_service(cls, auth: AuthSchema, search: Optional[PositionQueryParam] = None, order_by: Optional[List[Dict[str, str]]] = None) -> List[Dict]:
|
||||
"""获取岗位列表"""
|
||||
if order_by:
|
||||
order_by = eval(order_by)
|
||||
else:
|
||||
order_by = [{"order": "asc"}]
|
||||
position_list = await PositionCRUD(auth).get_list_crud(search=search.__dict__, order_by=order_by)
|
||||
return [PositionOutSchema.model_validate(position).model_dump() for position in position_list]
|
||||
|
||||
|
||||
@@ -30,7 +30,10 @@ async def get_obj_list_controller(
|
||||
search: RoleQueryParam = Depends(),
|
||||
auth: AuthSchema = Depends(AuthPermission(permissions=["system:role:query"])),
|
||||
) -> JSONResponse:
|
||||
result_dict_list = await RoleService.get_role_list_service(search=search, auth=auth, order_by=page.order_by)
|
||||
order_by = [{"order": "asc"}]
|
||||
if page.order_by:
|
||||
order_by = page.order_by
|
||||
result_dict_list = await RoleService.get_role_list_service(search=search, auth=auth, order_by=order_by)
|
||||
result_dict = await PaginationService.paginate(data_list= result_dict_list, page_no= page.page_no, page_size = page.page_size)
|
||||
logger.info(f"查询角色成功")
|
||||
return SuccessResponse(data=result_dict, msg="查询角色成功")
|
||||
|
||||
@@ -21,7 +21,7 @@ class RoleCRUD(CRUDBase[RoleModel, RoleCreateSchema, RoleUpdateSchema]):
|
||||
"""根据id获取角色信息"""
|
||||
return await self.get(id=id)
|
||||
|
||||
async def get_list_crud(self, search: Dict = None, order_by: List[Dict[str, str]] = None) -> Sequence[RoleModel]:
|
||||
async def get_list_crud(self, search: Optional[Dict] = None, order_by: Optional[List[Dict[str, str]]] = None) -> Sequence[RoleModel]:
|
||||
"""获取角色列表"""
|
||||
return await self.list(search=search, order_by=order_by)
|
||||
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
|
||||
from typing import Any, Dict, List
|
||||
from typing import Any, Dict, List, Optional
|
||||
|
||||
from app.core.base_schema import BatchSetAvailable
|
||||
from app.core.exceptions import CustomException
|
||||
@@ -26,12 +26,8 @@ class RoleService:
|
||||
return RoleOutSchema.model_validate(role).model_dump()
|
||||
|
||||
@classmethod
|
||||
async def get_role_list_service(cls, auth: AuthSchema, search: RoleQueryParam, order_by: List[Dict[str, str]] = None) -> List[Dict]:
|
||||
async def get_role_list_service(cls, auth: AuthSchema, search: Optional[RoleQueryParam] = None, order_by: Optional[List[Dict[str, str]]] = None) -> List[Dict]:
|
||||
"""获取角色列表"""
|
||||
if order_by:
|
||||
order_by = eval(order_by)
|
||||
else:
|
||||
order_by = [{"order": "asc"}]
|
||||
role_list = await RoleCRUD(auth).get_list_crud(search=search.__dict__, order_by=order_by)
|
||||
return [RoleOutSchema.model_validate(role).model_dump() for role in role_list]
|
||||
|
||||
|
||||
@@ -69,7 +69,7 @@ async def change_current_user_password_controller(
|
||||
return SuccessResponse(data=result_dict, msg='修改密码成功, 请重新登录')
|
||||
|
||||
@UserRouter.put("/reset/password", summary="重置密码", description="重置密码")
|
||||
async def change_current_user_password_controller(
|
||||
async def reset_password_controller(
|
||||
data: ResetPasswordSchema,
|
||||
auth: AuthSchema = Depends(get_current_user)
|
||||
) -> JSONResponse:
|
||||
|
||||
@@ -56,7 +56,7 @@ class UserCRUD(CRUDBase[UserModel, UserCreateSchema, UserUpdateSchema]):
|
||||
"""
|
||||
return await self.get(mobile=mobile)
|
||||
|
||||
async def get_list_crud(self, search: Dict = None, order_by: List[Dict[str, str]] = None) -> Sequence[UserModel]:
|
||||
async def get_list_crud(self, search: Optional[Dict] = None, order_by: Optional[List[Dict[str, str]]] = None) -> Sequence[UserModel]:
|
||||
"""
|
||||
获取用户列表
|
||||
|
||||
|
||||
@@ -21,7 +21,6 @@ class UserQueryParam:
|
||||
end_time: Optional[DateTimeStr] = Query(None, description="结束时间", example="2023-12-31 23:59:59"),
|
||||
creator: Optional[int] = Query(None, description="创建人"),
|
||||
) -> None:
|
||||
super().__init__()
|
||||
|
||||
# 模糊查询字段
|
||||
self.username = ("like", username)
|
||||
@@ -36,6 +35,6 @@ class UserQueryParam:
|
||||
|
||||
# 时间范围查询
|
||||
if start_time and end_time:
|
||||
start_datetime = datetime.strptime(start_time, '%Y-%m-%d %H:%M:%S')
|
||||
end_datetime = datetime.strptime(end_time, '%Y-%m-%d %H:%M:%S')
|
||||
start_datetime = datetime.strptime(str(start_time), '%Y-%m-%d %H:%M:%S')
|
||||
end_datetime = datetime.strptime(str(end_time), '%Y-%m-%d %H:%M:%S')
|
||||
self.created_at = ("between", (start_datetime, end_datetime))
|
||||
|
||||
@@ -1,7 +1,7 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
|
||||
import io
|
||||
from typing import Any, Dict, List
|
||||
from typing import Any, Dict, List, Optional
|
||||
from fastapi import UploadFile
|
||||
import pandas as pd
|
||||
|
||||
@@ -52,9 +52,7 @@ class UserService:
|
||||
return UserOutSchema.model_validate(user).model_dump()
|
||||
|
||||
@classmethod
|
||||
async def get_user_list_service(cls, auth: AuthSchema, search: UserQueryParam, order_by: List[Dict]= None) -> List[Dict]:
|
||||
if order_by:
|
||||
order_by = eval(order_by)
|
||||
async def get_user_list_service(cls, auth: AuthSchema, search: Optional[UserQueryParam] = None, order_by: Optional[List[Dict[str, str]]] = None) -> List[Dict]:
|
||||
user_list = await UserCRUD(auth).get_list_crud(search=search.__dict__, order_by=order_by)
|
||||
user_dict_list = []
|
||||
for user in user_list:
|
||||
|
||||
Reference in New Issue
Block a user