From ba7fddc34af5180418539ac3695a0a36a41cc658 Mon Sep 17 00:00:00 2001 From: zhangtao <9480807882@qq.com> Date: Wed, 1 Oct 2025 16:58:49 +0800 Subject: [PATCH] =?UTF-8?q?refactor(generator):=20=E4=BC=98=E5=8C=96?= =?UTF-8?q?=E4=BB=A3=E7=A0=81=E7=94=9F=E6=88=90=E6=A8=A1=E5=9D=97=E6=8E=A5?= =?UTF-8?q?=E5=8F=A3=E5=92=8C=E6=9C=8D=E5=8A=A1=E5=AE=9E=E7=8E=B0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 修改接口定义，增加路径参数并完善请求描述，增强参数校验和依赖注入 - 优化CRUD层数据库操作，统一异步会话使用，删除多余db参数 - 增加业务表与字段模型关系级联删除配置，优化模型关联关系声明 - 精简pydantic模型，去除冗余校验装饰器，完善字段描述和必填约束 - 服务层增加类型检查和异常抛出，规范业务逻辑流程和错误提示 - 优化代码结构，调整模块导入顺序和注释，提升代码可读性和一致性 --- .../api/v1/module_application/myapp/crud.py | 2 +- .../api/v1/module_example/demo/controller.py | 15 +- .../app/api/v1/module_example/demo/crud.py | 3 +- .../app/api/v1/module_example/demo/param.py | 1 - .../app/api/v1/module_example/demo/service.py | 75 +-- .../v1/module_generator/gencode/controller.py | 48 +- .../api/v1/module_generator/gencode/crud.py | 50 +- .../api/v1/module_generator/gencode/model.py | 25 +- .../api/v1/module_generator/gencode/param.py | 12 +- .../api/v1/module_generator/gencode/schema.py | 98 +--- .../v1/module_generator/gencode/service.py | 492 +++++++++--------- backend/app/api/v1/module_monitor/job/crud.py | 4 +- backend/app/api/v1/module_system/dept/crud.py | 6 +- backend/app/api/v1/module_system/dict/crud.py | 4 +- backend/app/api/v1/module_system/log/crud.py | 2 +- backend/app/api/v1/module_system/menu/crud.py | 6 +- .../app/api/v1/module_system/menu/model.py | 2 +- .../app/api/v1/module_system/notice/crud.py | 2 +- .../app/api/v1/module_system/params/crud.py | 2 +- .../api/v1/module_system/params/service.py | 12 +- .../v1/module_system/position/controller.py | 5 +- .../app/api/v1/module_system/position/crud.py | 2 +- .../api/v1/module_system/position/param.py | 4 +- .../api/v1/module_system/position/service.py | 8 +- .../api/v1/module_system/role/controller.py | 5 +- backend/app/api/v1/module_system/role/crud.py | 2 +- .../app/api/v1/module_system/role/service.py | 8 +- .../api/v1/module_system/user/controller.py | 2 +- backend/app/api/v1/module_system/user/crud.py | 2 +- .../app/api/v1/module_system/user/param.py | 5 +- .../app/api/v1/module_system/user/service.py | 6 +- backend/app/common/request.py | 2 +- backend/app/core/base_crud.py | 147 ++++-- backend/app/core/base_params.py | 4 +- backend/app/core/serialize.py | 71 +++ 35 files changed, 602 insertions(+), 532 deletions(-) create mode 100644 backend/app/core/serialize.py diff --git a/backend/app/api/v1/module_application/myapp/crud.py b/backend/app/api/v1/module_application/myapp/crud.py index 24098bc6..3bc8e248 100644 --- a/backend/app/api/v1/module_application/myapp/crud.py +++ b/backend/app/api/v1/module_application/myapp/crud.py @@ -20,7 +20,7 @@ class ApplicationCRUD(CRUDBase[ApplicationModel, ApplicationCreateSchema, Applic """获取应用详情""" return await self.get(id=id) - async def get_list_crud(self, search: Dict = None, order_by: List[Dict[str, str]] = None) -> Sequence[ApplicationModel]: + async def get_list_crud(self, search: Optional[Dict] = None, order_by: Optional[List[Dict[str, str]]] = None) -> Sequence[ApplicationModel]: """列表查询""" return await self.list(search=search, order_by=order_by) diff --git a/backend/app/api/v1/module_example/demo/controller.py b/backend/app/api/v1/module_example/demo/controller.py index 3b6c94ef..1a9ad1db 100644 --- a/backend/app/api/v1/module_example/demo/controller.py +++ b/backend/app/api/v1/module_example/demo/controller.py @@ -3,6 +3,7 @@ from fastapi import APIRouter, Body, Depends, Path, UploadFile from fastapi.responses import JSONResponse, StreamingResponse import urllib.parse +import json from app.common.response import StreamResponse, SuccessResponse from app.common.request import PaginationService @@ -39,8 +40,8 @@ async def get_obj_list_controller( auth: AuthSchema = Depends(AuthPermission(permissions=["demo:example:query"])) ) -> JSONResponse: result_dict_list = await DemoService.get_demo_list_service(auth=auth, search=search, order_by=page.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"查询示例列表成功") + result_dict = await PaginationService.paginate(data_list=result_dict_list, page_no=page.page_no, page_size=page.page_size) + logger.info("查询示例列表成功") return SuccessResponse(data=result_dict, msg="查询示例列表成功") @DemoRouter.post("/create", summary="创建示例", description="创建示例") @@ -49,7 +50,7 @@ async def create_obj_controller( auth: AuthSchema = Depends(AuthPermission(permissions=["demo:example:create"])) ) -> JSONResponse: result_dict = await DemoService.create_demo_service(auth=auth, data=data) - logger.info(f"创建示例成功: {result_dict}") + logger.info(f"创建示例成功: {result_dict.get('name')}") return SuccessResponse(data=result_dict, msg="创建示例成功") @DemoRouter.put("/update/{id}", summary="修改示例", description="修改示例") @@ -59,7 +60,7 @@ async def update_obj_controller( auth: AuthSchema = Depends(AuthPermission(permissions=["demo:example:update"])) ) -> JSONResponse: result_dict = await DemoService.update_demo_service(auth=auth, id=id, data=data) - logger.info(f"修改示例成功: {result_dict}") + logger.info(f"修改示例成功: {result_dict.get('name')}") return SuccessResponse(data=result_dict, msg="修改示例成功") @DemoRouter.delete("/delete", summary="删除示例", description="删除示例") @@ -93,7 +94,7 @@ async def export_obj_list_controller( return StreamResponse( data=bytes2file_response(export_result), media_type='application/vnd.openxmlformats-officedocument.spreadsheetml.sheet', - headers = { + headers={ 'Content-Disposition': 'attachment; filename=example.xlsx' } ) @@ -108,14 +109,14 @@ async def import_obj_list_controller( return SuccessResponse(data=batch_import_result, msg="导入示例成功") @DemoRouter.post('/download/template', summary="获取示例导入模板", description="获取示例导入模板", dependencies=[Depends(AuthPermission(permissions=["demo:example:download"]))]) -async def export_obj_template_controller()-> StreamingResponse: +async def export_obj_template_controller() -> StreamingResponse: example_import_template_result = await DemoService.import_template_download_service() logger.info('获取示例导入模板成功') return StreamResponse( data=bytes2file_response(example_import_template_result), media_type='application/vnd.openxmlformats-officedocument.spreadsheetml.sheet', - headers = { + headers={ 'Content-Disposition': f'attachment; filename={urllib.parse.quote("示例导入模板.xlsx")}', 'Access-Control-Expose-Headers': 'Content-Disposition' } diff --git a/backend/app/api/v1/module_example/demo/crud.py b/backend/app/api/v1/module_example/demo/crud.py index cd77b9fa..31ed9909 100644 --- a/backend/app/api/v1/module_example/demo/crud.py +++ b/backend/app/api/v1/module_example/demo/crud.py @@ -13,14 +13,13 @@ class DemoCRUD(CRUDBase[DemoModel, DemoCreateSchema, DemoUpdateSchema]): def __init__(self, auth: AuthSchema) -> None: """初始化CRUD""" - self.auth = auth super().__init__(model=DemoModel, auth=auth) async def get_by_id_crud(self, id: int) -> Optional[DemoModel]: """详情""" return await self.get(id=id) - async def get_list_crud(self, search: Dict = None, order_by: List[Dict[str, str]] = None) -> Sequence[DemoModel]: + async def get_list_crud(self, search: Optional[Dict] = None, order_by: Optional[List[Dict[str, str]]] = None) -> Sequence[DemoModel]: """列表查询""" return await self.list(search=search, order_by=order_by) diff --git a/backend/app/api/v1/module_example/demo/param.py b/backend/app/api/v1/module_example/demo/param.py index af135189..ef96acee 100644 --- a/backend/app/api/v1/module_example/demo/param.py +++ b/backend/app/api/v1/module_example/demo/param.py @@ -17,7 +17,6 @@ class DemoQueryParam: start_time: Optional[DateTimeStr] = Query(None, description="开始时间", example="2023-01-01 00:00:00"), end_time: Optional[DateTimeStr] = Query(None, description="结束时间", example="2023-12-31 23:59:59"), ) -> None: - super().__init__() # 模糊查询字段 self.name = ("like", name) diff --git a/backend/app/api/v1/module_example/demo/service.py b/backend/app/api/v1/module_example/demo/service.py index bf17476e..7230d487 100644 --- a/backend/app/api/v1/module_example/demo/service.py +++ b/backend/app/api/v1/module_example/demo/service.py @@ -1,7 +1,7 @@ # -*- coding: utf-8 -*- import io -from typing import Any, List, Dict +from typing import Any, List, Dict, Optional from fastapi import UploadFile import pandas as pd @@ -25,14 +25,15 @@ class DemoService: async def get_demo_detail_service(cls, auth: AuthSchema, id: int) -> Dict: """详情""" obj = await DemoCRUD(auth).get_by_id_crud(id=id) + if not obj: + raise CustomException(msg="该数据不存在") return DemoOutSchema.model_validate(obj).model_dump() @classmethod - async def get_demo_list_service(cls, auth: AuthSchema, search: DemoQueryParam = None, order_by: List[Dict[str, str]] = None) -> List[Dict]: + async def get_demo_list_service(cls, auth: AuthSchema, search: Optional[DemoQueryParam] = None, order_by: Optional[List[Dict[str, str]]] = None) -> List[Dict]: """列表查询""" - if order_by: - order_by = eval(order_by) - obj_list = await DemoCRUD(auth).get_list_crud(search=search.__dict__, order_by=order_by) + search_dict = search.__dict__ if search else None + obj_list = await DemoCRUD(auth).get_list_crud(search=search_dict, order_by=order_by) return [DemoOutSchema.model_validate(obj).model_dump() for obj in obj_list] @classmethod @@ -45,26 +46,33 @@ class DemoService: return DemoOutSchema.model_validate(obj).model_dump() @classmethod - async def update_demo_service(cls, auth: AuthSchema, id:int, data: DemoUpdateSchema) -> Dict: + async def update_demo_service(cls, auth: AuthSchema, id: int, data: DemoUpdateSchema) -> Dict: """更新""" + # 检查数据是否存在 obj = await DemoCRUD(auth).get_by_id_crud(id=id) if not obj: raise CustomException(msg='更新失败，该数据不存在') + + # 检查名称是否重复 exist_obj = await DemoCRUD(auth).get(name=data.name) - if exist_obj and exist_obj.id != data.id: + if exist_obj and exist_obj.id != id: raise CustomException(msg='更新失败，名称重复') + obj = await DemoCRUD(auth).update_crud(id=id, data=data) return DemoOutSchema.model_validate(obj).model_dump() @classmethod - async def delete_demo_service(cls, auth: AuthSchema, ids: list[int]) -> None: + async def delete_demo_service(cls, auth: AuthSchema, ids: List[int]) -> None: """删除""" if len(ids) < 1: raise CustomException(msg='删除失败，删除对象不能为空') + + # 检查所有要删除的数据是否存在 for id in ids: obj = await DemoCRUD(auth).get_by_id_crud(id=id) if not obj: - raise CustomException(msg='删除失败，该数据不存在') + raise CustomException(msg=f'删除失败，ID为{id}的数据不存在') + await DemoCRUD(auth).delete_crud(ids=ids) @classmethod @@ -90,10 +98,14 @@ class DemoService: for item in data: # 处理状态 item['status'] = '正常' if item.get('status') else '停用' - # 处理公告类型 - item['creator'] = item.get('creator', {}).get('name', '未知') if isinstance(item.get('creator'), dict) else '未知' + # 处理创建者 + creator_info = item.get('creator') + if isinstance(creator_info, dict): + item['creator'] = creator_info.get('name', '未知') + else: + item['creator'] = '未知' - return ExcelUtil.export_list2excel(list_data=obj_list, mapping_dict=mapping_dict) + return ExcelUtil.export_list2excel(list_data=data, mapping_dict=mapping_dict) @classmethod async def batch_import_service(cls, auth: AuthSchema, file: UploadFile, update_support: bool = False) -> str: @@ -125,9 +137,10 @@ class DemoService: # 验证必填字段 required_fields = ['name', 'status'] for field in required_fields: - if df[field].isnull().any(): - missing_rows = df[df[field].isnull()].index.tolist() - raise CustomException(msg=f"{[k for k,v in header_dict.items() if v == field][0]}不能为空，第{[i+1 for i in missing_rows]}行") + if df[field].isnull().any(): # type: ignore + missing_rows = df[df[field].isnull()].index.tolist() # type: ignore + row_numbers = [str(int(i)+2) for i in missing_rows] # +2 because index starts at 0 and first row is header + raise CustomException(msg=f"{[k for k,v in header_dict.items() if v == field][0]}不能为空，第{', '.join(row_numbers)}行") error_msgs = [] success_count = 0 @@ -139,35 +152,37 @@ class DemoService: try: name = str(row['name']) except ValueError: - error_msgs.append(f"第{index+1}行: 名称必须是字符串") - continue - try: - status = True if row['status'] == '正常' else False - except ValueError: - error_msgs.append(f"第{index+1}行: 状态必须是'正常'或'停用'") + error_msgs.append(f"第{int(index)+2}行: 名称必须是字符串") continue - # 构建用户数据 + # 处理状态字段 + status_value = row['status'] + if isinstance(status_value, str): + status = status_value == '正常' + else: + status = bool(status_value) + + # 构建数据 data = { "name": name, "status": status, "description": str(row['description']).strip() if not pd.isna(row['description']) else None, } - # 处理用户导入 - exists_user = await DemoCRUD(auth).get(name=data["name"]) - if exists_user: + # 处理导入逻辑 + exists_obj = await DemoCRUD(auth).get(name=data["name"]) + if exists_obj: if update_support: - await DemoCRUD(auth).update(id=exists_user.id, data=data) + await DemoCRUD(auth).update_crud(id=exists_obj.id, data=DemoUpdateSchema(**data)) success_count += 1 else: - error_msgs.append(f"第{index+1}行: 用户 {data['username']} 已存在") + error_msgs.append(f"第{int(index)+2}行: 名称 {data['name']} 已存在") else: - await DemoCRUD(auth).create(data=data) + await DemoCRUD(auth).create_crud(data=DemoCreateSchema(**data)) success_count += 1 except Exception as e: - error_msgs.append(f"第{index+1}行: {str(e)}") + error_msgs.append(f"第{int(index)+2}行: {str(e)}") continue # 返回详细的导入结果 @@ -177,7 +192,7 @@ class DemoService: return result except Exception as e: - logger.error(f"批量导入用户失败: {str(e)}") + logger.error(f"批量导入失败: {str(e)}") raise CustomException(msg=f"导入失败: {str(e)}") @classmethod diff --git a/backend/app/api/v1/module_generator/gencode/controller.py b/backend/app/api/v1/module_generator/gencode/controller.py index 3d11ae43..596c710a 100644 --- a/backend/app/api/v1/module_generator/gencode/controller.py +++ b/backend/app/api/v1/module_generator/gencode/controller.py @@ -2,9 +2,9 @@ from datetime import datetime from typing import List -from fastapi import APIRouter, Depends, Query, Body +from fastapi import APIRouter, Depends, Query, Body, Path from fastapi.responses import StreamingResponse, JSONResponse -from pydantic_validation_decorator import ValidateFields + from app.common.response import SuccessResponse, ErrorResponse, StreamResponse from app.core.dependencies import AuthPermission from app.core.router_class import OperationLogRoute @@ -22,9 +22,9 @@ from app.core.logger import logger GenRouter = APIRouter(route_class=OperationLogRoute, prefix='/gencode', tags=["代码生成模块"]) -@GenRouter.get('/detail/{table_id}', summary="获取业务表详细信息", description="获取业务表详细信息") -async def query_detail_gen_table_controller( - table_id: int, +@GenRouter.get("/detail/{table_id}", summary="获取业务表详细信息", description="获取业务表详细信息") +async def get_gen_table_detail_controller( + table_id: int = Path(..., description="业务表ID"), auth: AuthSchema = Depends(AuthPermission(permissions=["generator:gencode:query"])) ) -> JSONResponse: gen_table = await GenTableService.get_gen_table_by_id_service(auth, table_id) @@ -35,7 +35,7 @@ async def query_detail_gen_table_controller( return SuccessResponse(data=gen_table_detail_result, msg="获取业务表详细信息成功") -@GenRouter.get('/list', summary="查询代码生成业务表列表", description="查询代码生成业务表列表") +@GenRouter.get("/list", summary="查询代码生成业务表列表", description="查询代码生成业务表列表") async def get_gen_table_list_controller( page: PaginationQueryParam = Depends(), search: GenTableQueryParam = Depends(), @@ -46,7 +46,8 @@ async def get_gen_table_list_controller( logger.info('获取代码生成业务表列表成功') return SuccessResponse(data=result_dict, msg="获取代码生成业务表列表成功") -@GenRouter.post('/create', summary="创建表结构", description="创建表结构") + +@GenRouter.post("/create", summary="创建表结构", description="创建表结构") async def create_table_controller( sql: str = Query(..., description="SQL语句"), auth: AuthSchema = Depends(AuthPermission(permissions=["generator:gencode:create"])), @@ -57,10 +58,10 @@ async def create_table_controller( return SuccessResponse(msg="创建表结构成功", data=result) -@GenRouter.put('/update', summary="编辑业务表信息", description="编辑业务表信息") -@ValidateFields(validate_model='edit_gen_table') +@GenRouter.put("/update/{table_id}", summary="编辑业务表信息", description="编辑业务表信息") async def update_gen_table_controller( - data: GenTableUpdateSchema, + table_id: int = Path(..., description="业务表ID"), + data: GenTableUpdateSchema = Body(..., description="业务表信息"), auth: AuthSchema = Depends(AuthPermission(permissions=["generator:gencode:update"])), current_user: UserOutSchema = Depends(lambda auth: auth.user) ) -> JSONResponse: @@ -73,14 +74,14 @@ async def update_gen_table_controller( updated_data = GenTableUpdateSchema(**update_data) await GenTableService.validate_edit(updated_data) - edit_gen_result = await GenTableService.edit_gen_table_service(auth, updated_data) + edit_gen_result = await GenTableService.update_gen_table_service(auth, updated_data, table_id) logger.info('编辑业务表信息成功') return SuccessResponse(data=edit_gen_result, msg="编辑业务表信息成功") -@GenRouter.delete('/delete', summary="删除业务表信息", description="删除业务表信息") +@GenRouter.delete("/delete", summary="删除业务表信息", description="删除业务表信息") async def delete_gen_table_controller( - table_ids: list[int] = Body(..., description="ID列表"), + table_ids: str = Body(..., description="ID列表，用逗号分隔"), auth: AuthSchema = Depends(AuthPermission(permissions=["generator:gencode:delete"])) ) -> JSONResponse: delete_gen_table = GenTableDeleteSchema(table_ids=table_ids) @@ -89,8 +90,7 @@ async def delete_gen_table_controller( return result -@GenRouter.post('/import', summary="导入表结构", description="导入表结构") -@ValidateFields(validate_model='edit_gen_table') +@GenRouter.post("/import", summary="导入表结构", description="导入表结构") async def import_gen_table_controller( tables: List[str] = Body(..., description="表名列表", embed=True), auth: AuthSchema = Depends(AuthPermission(permissions=["generator:gencode:import"])), @@ -103,9 +103,9 @@ async def import_gen_table_controller( return result -@GenRouter.patch('/batch/out', summary="批量生成代码", description="批量生成代码") +@GenRouter.patch("/batch/out", summary="批量生成代码", description="批量生成代码") async def batch_gen_code_controller( - tables: str = Query(..., description="表名列表"), + tables: str = Query(..., description="表名列表，用逗号分隔"), auth: AuthSchema = Depends(AuthPermission(permissions=["generator:gencode:operate"])) ) -> StreamResponse: table_names = tables.split(',') if tables else [] @@ -118,9 +118,9 @@ async def batch_gen_code_controller( ) -@GenRouter.post('/out/path/{table_name}', summary="生成代码到指定路径", description="生成代码到指定路径") +@GenRouter.post("/out/path/{table_name}", summary="生成代码到指定路径", description="生成代码到指定路径") async def gen_code_local_controller( - table_name: str, + table_name: str = Path(..., description="表名"), auth: AuthSchema = Depends(AuthPermission(permissions=["generator:gencode:code"])) ) -> JSONResponse: from app.config.setting import settings @@ -132,9 +132,9 @@ async def gen_code_local_controller( return SuccessResponse(msg="生成代码到指定路径成功", data=result) -@GenRouter.get('/preview/{table_id}', summary="预览代码", description="预览代码") +@GenRouter.get("/preview/{table_id}", summary="预览代码", description="预览代码") async def preview_code_controller( - table_id: int, + table_id: int = Path(..., description="业务表ID"), auth: AuthSchema = Depends(AuthPermission(permissions=["generator:gencode:query"])) ) -> JSONResponse: preview_code_result = await GenTableService.preview_code_service(auth, table_id) @@ -142,7 +142,7 @@ async def preview_code_controller( return SuccessResponse(data=preview_code_result, msg="预览代码成功") -@GenRouter.get('/db/list', summary="查询数据库表列表", description="查询数据库表列表") +@GenRouter.get("/db/list", summary="查询数据库表列表", description="查询数据库表列表") async def get_gen_db_table_list_controller( page: PaginationQueryParam = Depends(), search: GenTableQueryParam = Depends(), @@ -154,9 +154,9 @@ async def get_gen_db_table_list_controller( return SuccessResponse(data=result_dict, msg="获取数据库表列表成功") -@GenRouter.post('/sync/db/{table_name}', summary="同步数据库", description="同步数据库") +@GenRouter.post("/sync/db/{table_name}", summary="同步数据库", description="同步数据库") async def sync_db_controller( - table_name: str, + table_name: str = Path(..., description="表名"), auth: AuthSchema = Depends(AuthPermission(permissions=["generator:db:sync"])) ) -> JSONResponse: result = await GenTableService.sync_db_service(auth, table_name) diff --git a/backend/app/api/v1/module_generator/gencode/crud.py b/backend/app/api/v1/module_generator/gencode/crud.py index 2ffd53b3..97d3c61f 100644 --- a/backend/app/api/v1/module_generator/gencode/crud.py +++ b/backend/app/api/v1/module_generator/gencode/crud.py @@ -17,15 +17,13 @@ from app.api.v1.module_system.auth.schema import AuthSchema class GenTableCRUD(CRUDBase[GenTableModel, GenTableCreateSchema, GenTableUpdateSchema]): - """ - 代码生成业务表模块数据库操作层 - """ + """代码生成业务表模块数据库操作层""" def __init__(self, auth: AuthSchema) -> None: """初始化CRUD""" super().__init__(model=GenTableModel, auth=auth) - async def get_gen_table_by_id(self, db: AsyncSession, table_id: int) -> Optional[GenTableModel]: + async def get_gen_table_by_id(self, table_id: int) -> Optional[GenTableModel]: """ 根据业务表id获取需要生成的业务表信息 @@ -35,7 +33,7 @@ class GenTableCRUD(CRUDBase[GenTableModel, GenTableCreateSchema, GenTableUpdateS """ gen_table_info = ( ( - await db.execute( + await self.db.execute( select(GenTableModel).options(selectinload(GenTableModel.columns)).where(GenTableModel.id == table_id) ) ) @@ -45,7 +43,7 @@ class GenTableCRUD(CRUDBase[GenTableModel, GenTableCreateSchema, GenTableUpdateS return gen_table_info - async def get_gen_table_by_name(self, db: AsyncSession, table_name: str) -> Optional[GenTableModel]: + async def get_gen_table_by_name(self, table_name: str) -> Optional[GenTableModel]: """ 根据业务表名称获取需要生成的业务表信息 @@ -55,7 +53,7 @@ class GenTableCRUD(CRUDBase[GenTableModel, GenTableCreateSchema, GenTableUpdateS """ gen_table_info = ( ( - await db.execute( + await self.db.execute( select(GenTableModel).options(selectinload(GenTableModel.columns)).where(GenTableModel.table_name == table_name) ) ) @@ -65,18 +63,18 @@ class GenTableCRUD(CRUDBase[GenTableModel, GenTableCreateSchema, GenTableUpdateS return gen_table_info - async def get_gen_table_all(self, db: AsyncSession) -> Sequence[GenTableModel]: + async def get_gen_table_all(self) -> Sequence[GenTableModel]: """ 获取所有业务表信息 :param db: orm对象 :return: 所有业务表信息 """ - gen_table_all = (await db.execute(select(GenTableModel).options(selectinload(GenTableModel.columns)))).scalars().all() + gen_table_all = (await self.db.execute(select(GenTableModel).options(selectinload(GenTableModel.columns)))).scalars().all() return gen_table_all - async def create_table_by_sql_dao(self, db: AsyncSession, sql_statements: List) -> None: + async def create_table_by_sql(self, sql_statements: List) -> None: """ 根据sql语句创建表结构 @@ -86,9 +84,9 @@ class GenTableCRUD(CRUDBase[GenTableModel, GenTableCreateSchema, GenTableUpdateS """ for sql_statement in sql_statements: sql = sql_statement.sql(dialect=settings.DATABASE_TYPE) - await db.execute(text(sql)) + await self.db.execute(text(sql)) - async def get_gen_table_list(self, db: AsyncSession, query_object: GenTableQueryParam, is_page: bool = False): + async def get_gen_table_list(self, query_object: GenTableQueryParam, is_page: bool = False): """ 根据查询参数获取代码生成业务表列表信息 @@ -122,15 +120,18 @@ class GenTableCRUD(CRUDBase[GenTableModel, GenTableCreateSchema, GenTableUpdateS ) # 获取所有数据 - result = await db.execute(query) + result = await self.db.execute(query) all_data = list(result.scalars().all()) # 使用PaginationService.paginate进行分页 - if is_page and query_object.page_no is not None and query_object.page_size is not None: + # 注意：这里假设query_object有page_no和page_size属性，如果没有需要从其他地方获取 + page_no = getattr(query_object, 'page_no', None) + page_size = getattr(query_object, 'page_size', None) + if is_page and page_no is not None and page_size is not None: paginated_result = await PaginationService.paginate( data_list=all_data, - page_no=query_object.page_no, - page_size=query_object.page_size + page_no=page_no, + page_size=page_size ) return paginated_result else: @@ -191,11 +192,14 @@ class GenTableCRUD(CRUDBase[GenTableModel, GenTableCreateSchema, GenTableUpdateS all_data = list(result.fetchall()) # 使用PaginationService.paginate进行分页 - if is_page and query_object.page_no is not None and query_object.page_size is not None: + # 注意：这里假设query_object有page_no和page_size属性，如果没有需要从其他地方获取 + page_no = getattr(query_object, 'page_no', None) + page_size = getattr(query_object, 'page_size', None) + if is_page and page_no is not None and page_size is not None: paginated_result = await PaginationService.paginate( data_list=all_data, - page_no=query_object.page_no, - page_size=query_object.page_size + page_no=page_no, + page_size=page_size ) return paginated_result else: @@ -234,14 +238,16 @@ class GenTableCRUD(CRUDBase[GenTableModel, GenTableCreateSchema, GenTableUpdateS class GenTableColumnCRUD(CRUDBase[GenTableColumnModel, GenTableColumnCreateSchema, GenTableColumnUpdateSchema]): - """ - 代码生成业务表字段模块数据库操作层 - """ + """代码生成业务表字段模块数据库操作层""" def __init__(self, auth: AuthSchema) -> None: """初始化CRUD""" super().__init__(model=GenTableColumnModel, auth=auth) + async def get_gen_table_column_list_by_table_id_crud(self, table_id: int) -> Sequence[GenTableColumnModel]: + """根据业务表id获取需要生成的业务表字段列表信息""" + return await self.list(search={"table_id": table_id}) + async def get_gen_table_column_list_by_table_id(self, db: AsyncSession, table_id: int) -> Sequence[GenTableColumnModel]: """ 根据业务表id获取需要生成的业务表字段列表信息 diff --git a/backend/app/api/v1/module_generator/gencode/model.py b/backend/app/api/v1/module_generator/gencode/model.py index 2da89f68..c0ae71e0 100644 --- a/backend/app/api/v1/module_generator/gencode/model.py +++ b/backend/app/api/v1/module_generator/gencode/model.py @@ -2,7 +2,7 @@ from typing import Optional, List from sqlalchemy import String, Integer, ForeignKey -from sqlalchemy.orm import Mapped, mapped_column, relationship +from sqlalchemy.orm import Mapped, mapped_column, relationship, declared_attr from app.core.base_model import CreatorMixin @@ -30,7 +30,13 @@ class GenTableModel(CreatorMixin): gen_path: Mapped[Optional[str]] = mapped_column(String(200), nullable=True, default='/', comment='生成路径（不填默认项目路径）') options: Mapped[Optional[str]] = mapped_column(String(1000), nullable=True, comment='其它生成选项') - columns: Mapped[List['GenTableColumnModel']] = relationship('GenTableColumnModel', order_by='GenTableColumnModel.sort', back_populates='table') + # 关系定义 + columns: Mapped[List['GenTableColumnModel']] = relationship( + 'GenTableColumnModel', + order_by='GenTableColumnModel.sort', + back_populates='table', + cascade='all, delete-orphan' + ) class GenTableColumnModel(CreatorMixin): @@ -58,5 +64,16 @@ class GenTableColumnModel(CreatorMixin): dict_type: Mapped[Optional[str]] = mapped_column(String(200), nullable=True, default='', comment='字典类型') sort: Mapped[Optional[int]] = mapped_column(Integer, nullable=True, comment='排序') - table_id: Mapped[Optional[int]] = mapped_column(Integer, ForeignKey('gen_table.id'), nullable=True, comment='归属表编号') - table: Mapped['GenTableModel'] = relationship('GenTableModel', back_populates='columns') \ No newline at end of file + # 外键关系 + table_id: Mapped[Optional[int]] = mapped_column( + Integer, + ForeignKey('gen_table.id', ondelete='CASCADE'), + nullable=True, + comment='归属表编号' + ) + + # 关系定义 + table: Mapped['GenTableModel'] = relationship( + 'GenTableModel', + back_populates='columns' + ) \ No newline at end of file diff --git a/backend/app/api/v1/module_generator/gencode/param.py b/backend/app/api/v1/module_generator/gencode/param.py index 057c7dbf..98cd40f4 100644 --- a/backend/app/api/v1/module_generator/gencode/param.py +++ b/backend/app/api/v1/module_generator/gencode/param.py @@ -8,7 +8,7 @@ from app.core.validator import DateTimeStr class GenTableQueryParam: - """数据库表查询参数""" + """代码生成业务表查询参数""" def __init__( self, @@ -17,8 +17,8 @@ class GenTableQueryParam: creator: Optional[int] = Query(None, description="创建人"), start_time: Optional[DateTimeStr] = Query(None, description="开始时间", example="2023-01-01 00:00:00"), end_time: Optional[DateTimeStr] = Query(None, description="结束时间", example="2023-12-31 23:59:59"), - ) -> None: - # 存储查询条件，不直接赋值给父类属性 + ) -> None: + # 模糊查询字段 self.table_name = ("like", table_name) self.table_comment = ("like", table_comment) @@ -33,7 +33,7 @@ class GenTableQueryParam: class GenTableColumnQueryParam: - """数据库表字段查询参数""" + """代码生成业务表字段查询参数""" def __init__( self, @@ -41,8 +41,8 @@ class GenTableColumnQueryParam: creator: Optional[int] = Query(None, description="创建人"), start_time: Optional[DateTimeStr] = Query(None, description="开始时间", example="2023-01-01 00:00:00"), end_time: Optional[DateTimeStr] = Query(None, description="结束时间", example="2023-12-31 23:59:59"), - ) -> None: - # 存储查询条件，不直接赋值给父类属性 + ) -> None: + # 模糊查询字段 self.column_name = ("like", column_name) # 精确查询字段 diff --git a/backend/app/api/v1/module_generator/gencode/schema.py b/backend/app/api/v1/module_generator/gencode/schema.py index 60b9c480..d5c9ff7b 100644 --- a/backend/app/api/v1/module_generator/gencode/schema.py +++ b/backend/app/api/v1/module_generator/gencode/schema.py @@ -3,7 +3,6 @@ from typing import List, Literal, Optional from pydantic import BaseModel, ConfigDict, Field, model_validator from pydantic.alias_generators import to_camel -from pydantic_validation_decorator import NotBlank from app.utils.string_util import StringUtil from app.common.constant import GenConstant @@ -12,76 +11,34 @@ from app.core.base_schema import BaseSchema class GenTableCreateSchema(BaseModel): """ - 代码生成业务表对应pydantic模型 + 代码生成业务表创建模型 """ model_config = ConfigDict(from_attributes=True) - table_name: str = Field(default=..., description='表名称') - table_comment: str = Field(default=..., description='表描述') + table_name: str = Field(..., description='表名称') + table_comment: str = Field(..., description='表描述') sub_table_name: Optional[str] = Field(default=None, description='关联子表的表名') - sub_table_fk_name: str = Field(default=..., description='子表关联的外键名') - class_name: str = Field(default=..., description='实体类名称') + sub_table_fk_name: str = Field(..., description='子表关联的外键名') + class_name: str = Field(..., description='实体类名称') tpl_category: Optional[str] = Field(default=None, description='使用的模板（crud单表操作 tree树表操作）') tpl_web_type: Optional[str] = Field(default=None, description='前端模板类型（element-ui模版 element-plus模版）') - package_name: str = Field(default=..., description='生成包路径') - module_name: str = Field(default=..., description='生成模块名') - business_name: str = Field(default=..., description='生成业务名') - function_name: str = Field(default=..., description='生成功能名') + package_name: str = Field(..., description='生成包路径') + module_name: str = Field(..., description='生成模块名') + business_name: str = Field(..., description='生成业务名') + function_name: str = Field(..., description='生成功能名') function_author: Optional[str] = Field(default=None, description='生成功能作者') gen_type: Optional[Literal['0', '1']] = Field(default=None, description='生成代码方式（0zip压缩包 1自定义路径）') gen_path: Optional[str] = Field(default=None, description='生成路径（不填默认项目路径）') options: Optional[str] = Field(default=None, description='其它生成选项') - @NotBlank(field_name='table_name', message='表名称不能为空') - def get_table_name(self): - return self.table_name - - @NotBlank(field_name='table_comment', message='表描述不能为空') - def get_table_comment(self): - return self.table_comment - - @NotBlank(field_name='class_name', message='实体类名称不能为空') - def get_class_name(self): - return self.class_name - - @NotBlank(field_name='package_name', message='生成包路径不能为空') - def get_package_name(self): - return self.package_name - - @NotBlank(field_name='module_name', message='生成模块名不能为空') - def get_module_name(self): - return self.module_name - - @NotBlank(field_name='business_name', message='生成业务名不能为空') - def get_business_name(self): - return self.business_name - - @NotBlank(field_name='function_name', message='生成功能名不能为空') - def get_function_name(self): - return self.function_name - - @NotBlank(field_name='function_author', message='生成功能作者不能为空') - def get_function_author(self): - return self.function_author - - def validate_fields(self): - self.get_table_name() - self.get_table_comment() - self.get_class_name() - self.get_package_name() - self.get_module_name() - self.get_business_name() - self.get_function_name() - self.get_function_author() - class GenTableUpdateSchema(GenTableCreateSchema): """ - 代码生成业务表模型 + 代码生成业务表更新模型 """ pk_column: Optional['GenTableColumnUpdateSchema'] = Field(default=None, description='主键信息') sub_table: Optional['GenTableUpdateSchema'] = Field(default=None, description='子表信息') - columns: List['GenTableColumnUpdateSchema'] = Field(default=..., description='表列信息') + columns: List['GenTableColumnUpdateSchema'] = Field(..., description='表列信息') tree_code: Optional[str] = Field(default=None, description='树编码字段') tree_parent_code: Optional[str] = Field(default=None, description='树父编码字段') tree_name: Optional[str] = Field(default=None, description='树名称字段') @@ -100,7 +57,9 @@ class GenTableUpdateSchema(GenTableCreateSchema): class GenTableOutSchema(GenTableUpdateSchema, BaseSchema): - """响应模型""" + """ + 代码生成业务表响应模型 + """ model_config = ConfigDict(from_attributes=True) @@ -110,22 +69,22 @@ class GenTableDeleteSchema(BaseModel): """ model_config = ConfigDict(alias_generator=to_camel) - table_ids: str = Field(description='需要删除的代码生成业务表ID') + table_ids: str = Field(..., description='需要删除的代码生成业务表ID') class GenTableColumnCreateSchema(BaseModel): """ - 代码生成业务表字段对应pydantic模型 + 代码生成业务表字段创建模型 """ model_config = ConfigDict(alias_generator=to_camel, from_attributes=True) table_id: Optional[int] = Field(default=None, description='归属表编号') - column_name: str = Field(default=..., description='列名称') + column_name: str = Field(..., description='列名称') column_comment: Optional[str] = Field(default=None, description='列描述') - column_type: str = Field(default=..., description='列类型') + column_type: str = Field(..., description='列类型') python_type: Optional[str] = Field(default=None, description='PYTHON类型') - python_field: str = Field(default=..., description='PYTHON字段名') + python_field: str = Field(..., description='PYTHON字段名') is_pk: Optional[str] = Field(default=None, description='是否主键（1是）') is_increment: Optional[str] = Field(default=None, description='是否自增（1是）') is_required: Optional[str] = Field(default=None, description='是否必填（1是）') @@ -135,21 +94,14 @@ class GenTableColumnCreateSchema(BaseModel): is_list: Optional[str] = Field(default=None, description='是否列表字段（1是）') is_query: Optional[str] = Field(default=None, description='是否查询字段（1是）') query_type: Optional[str] = Field(default=None, description='查询方式（等于、不等于、大于、小于、范围）') - html_type: str = Field(default=..., description='显示类型（文本框、文本域、下拉框、复选框、单选框、日期控件）') - dict_type: str = Field(default=..., description='字典类型') + html_type: str = Field(..., description='显示类型（文本框、文本域、下拉框、复选框、单选框、日期控件）') + dict_type: str = Field(..., description='字典类型') sort: Optional[int] = Field(default=None, description='排序') - - @NotBlank(field_name='python_field', message='Python属性不能为空') - def get_python_field(self): - return self.python_field - - def validate_fields(self): - self.get_python_field() class GenTableColumnUpdateSchema(GenTableColumnCreateSchema): """ - 代码生成业务表字段模型 + 代码生成业务表字段更新模型 """ cap_python_field: Optional[str] = Field(default=None, description='字段大写形式') @@ -187,7 +139,9 @@ class GenTableColumnUpdateSchema(GenTableColumnCreateSchema): class GenTableColumnOutSchema(GenTableColumnUpdateSchema, BaseSchema): - """响应模型""" + """ + 代码生成业务表字段响应模型 + """ model_config = ConfigDict(from_attributes=True) @@ -197,4 +151,4 @@ class GenTableColumnDeleteSchema(BaseModel): """ model_config = ConfigDict(from_attributes=True) - column_ids: str = Field(description='需要删除的代码生成业务表字段ID') + column_ids: str = Field(..., description='需要删除的代码生成业务表字段ID') \ No newline at end of file diff --git a/backend/app/api/v1/module_generator/gencode/service.py b/backend/app/api/v1/module_generator/gencode/service.py index 328d4f7c..720b18e9 100644 --- a/backend/app/api/v1/module_generator/gencode/service.py +++ b/backend/app/api/v1/module_generator/gencode/service.py @@ -5,8 +5,8 @@ import json import os import zipfile from datetime import datetime -from sqlalchemy.ext.asyncio import AsyncSession from typing import Any, List, Dict, Optional, Sequence +from sqlalchemy.ext.asyncio import AsyncSession from app.config.setting import settings from app.core.base_model import CamelCaseUtil @@ -28,115 +28,139 @@ GEN_PATH = "generated_code" # 默认生成路径 class GenTableService: - """ - 代码生成业务表服务层 - """ + """代码生成业务表服务层""" + + @classmethod + async def get_gen_table_detail_service(cls, auth: AuthSchema, table_id: int) -> Dict: + """获取业务表详细信息""" + gen_table = await cls.get_gen_table_by_id_service(auth, table_id) + gen_tables = await cls.get_gen_table_all_service(auth) + gen_columns = await GenTableColumnService.get_gen_table_column_list_by_table_id_service(auth, table_id) + return dict(info=gen_table, rows=gen_columns, tables=gen_tables) @classmethod async def get_gen_table_list_service( cls, auth: AuthSchema, query_object: GenTableQueryParam, is_page: bool = False - ): - """ - 获取代码生成业务表列表信息service - - :param auth: 认证信息 - :param query_object: 查询参数对象 - :param is_page: 是否开启分页 - :return: 代码生成业务列表信息对象 - """ + ) -> Dict: + """获取代码生成业务表列表信息""" + if not auth.db: + raise CustomException(msg='数据库连接不存在') + # 确保db是AsyncSession类型 + db = auth.db + if not isinstance(db, AsyncSession): + raise CustomException(msg='数据库连接类型不正确') gen_table_dao = GenTableCRUD(auth=auth) - gen_table_list_result = await gen_table_dao.get_gen_table_list(auth.db, query_object, is_page) - + gen_table_list_result = await gen_table_dao.get_gen_table_list(db, query_object, is_page) return gen_table_list_result @classmethod async def get_gen_db_table_list_service( cls, auth: AuthSchema, query_object: GenTableQueryParam, is_page: bool = False - ): - """ - 获取数据库列表信息service - - :param auth: 认证信息 - :param query_object: 查询参数对象 - :param is_page: 是否开启分页 - :return: 数据库列表信息对象 - """ + ) -> Dict: + """获取数据库列表信息""" + if not auth.db: + raise CustomException(msg='数据库连接不存在') + # 确保db是AsyncSession类型 + db = auth.db + if not isinstance(db, AsyncSession): + raise CustomException(msg='数据库连接类型不正确') gen_table_dao = GenTableCRUD(auth=auth) - gen_db_table_list_result = await gen_table_dao.get_gen_db_table_list(auth.db, query_object, is_page) - + gen_db_table_list_result = await gen_table_dao.get_gen_db_table_list(db, query_object, is_page) return gen_db_table_list_result @classmethod - async def get_gen_db_table_list_by_name_service(cls, auth: AuthSchema, table_names: List[str]) -> list[GenTableOutSchema]: - """ - 根据表名称组获取数据库列表信息service - - :param auth: 认证信息 - :param table_names: 表名称组 - :return: 数据库列表信息对象 - """ + async def get_gen_db_table_list_by_name_service(cls, auth: AuthSchema, table_names: List[str]) -> List[GenTableOutSchema]: + """根据表名称组获取数据库列表信息""" + if not auth.db: + raise CustomException(msg='数据库连接不存在') + # 确保db是AsyncSession类型 + db = auth.db + if not isinstance(db, AsyncSession): + raise CustomException(msg='数据库连接类型不正确') gen_table_dao = GenTableCRUD(auth=auth) - gen_db_table_list_result = await gen_table_dao.get_gen_db_table_list_by_names(auth.db, table_names) - + gen_db_table_list_result = await gen_table_dao.get_gen_db_table_list_by_names(db, table_names) return [GenTableOutSchema(**gen_table) for gen_table in CamelCaseUtil.transform_result(gen_db_table_list_result)] @classmethod async def import_gen_table_service( cls, auth: AuthSchema, gen_table_list: List[GenTableOutSchema], current_user: UserOutSchema - ): - """ - 导入表结构service - - :param auth: 认证信息 - :param gen_table_list: 导入表列表 - :param current_user: 当前用户信息对象 - :return: 导入结果 - """ + ) -> SuccessResponse: + """导入表结构""" + if not auth.db: + raise CustomException(msg='数据库连接不存在') + # 确保db是AsyncSession类型 + db = auth.db + if not isinstance(db, AsyncSession): + raise CustomException(msg='数据库连接类型不正确') try: gen_table_dao = GenTableCRUD(auth=auth) gen_table_column_dao = GenTableColumnCRUD(auth=auth) for table in gen_table_list: table_name = table.table_name - GenUtils.init_table(table, current_user.username) # 使用username而不是user.user_name + GenUtils.init_table(table, current_user.username) add_gen_table = await gen_table_dao.create(data=table.model_dump()) if add_gen_table: - table.table_id = add_gen_table.id - gen_table_columns = await gen_table_column_dao.get_gen_db_table_columns_by_name(auth.db, table_name or "") + # 使用id而不是table_id + table.id = add_gen_table.id + gen_table_columns = await gen_table_column_dao.get_gen_db_table_columns_by_name(db, table_name or "") for column in [ GenTableColumnOutSchema(**gen_table_column) for gen_table_column in CamelCaseUtil.transform_result(gen_table_columns) ]: GenUtils.init_column_field(column, table) await gen_table_column_dao.create(data=column.model_dump()) - await auth.db.commit() + if isinstance(db, AsyncSession): + await db.commit() return SuccessResponse(msg='导入成功') except Exception as e: - try: - await auth.db.rollback() - except: - pass # 忽略回滚错误 + if isinstance(db, AsyncSession): + try: + await db.rollback() + except: + pass # 忽略回滚错误 raise CustomException(msg=f'导入失败, {str(e)}') @classmethod - async def edit_gen_table_service(cls, auth: AuthSchema, page_object: GenTableUpdateSchema) -> Dict[str, Any]: - """ - 编辑业务表信息service + async def create_table_service(cls, auth: AuthSchema, sql: str, current_user: UserOutSchema) -> SuccessResponse: + """创建表结构""" + if not auth.db: + raise CustomException(msg='数据库连接不存在') + # 确保db是AsyncSession类型 + db = auth.db + if not isinstance(db, AsyncSession): + raise CustomException(msg='数据库连接类型不正确') + gen_table_dao = GenTableCRUD(auth=auth) + + try: + # 执行SQL语句创建表 + await gen_table_dao.create_table_by_sql_dao(db, [sql]) + if isinstance(db, AsyncSession): + await db.commit() + return SuccessResponse(msg='创建表结构成功') + except Exception as e: + if isinstance(db, AsyncSession): + try: + await db.rollback() + except: + pass # 忽略回滚错误 + raise CustomException(msg=f'创建表结构失败: {str(e)}') - :param auth: 认证信息 - :param page_object: 编辑业务表对象 - :return: 编辑业务表校验结果 - """ + @classmethod + async def update_gen_table_service(cls, auth: AuthSchema, page_object: GenTableUpdateSchema, table_id: int) -> Dict[str, Any]: + """编辑业务表信息""" + if not auth.db: + raise CustomException(msg='数据库连接不存在') + # 确保db是AsyncSession类型 + db = auth.db + if not isinstance(db, AsyncSession): + raise CustomException(msg='数据库连接类型不正确') gen_table_dao = GenTableCRUD(auth=auth) gen_table_column_dao = GenTableColumnCRUD(auth=auth) - # 检查必要字段是否存在 - if getattr(page_object, 'table_id', None) is None: - raise CustomException(msg='业务表ID不能为空') - edit_gen_table = page_object.model_dump(exclude_unset=True, by_alias=True) - gen_table_info = await cls.get_gen_table_by_id_service(auth, page_object.table_id) - if gen_table_info.table_id: + gen_table_info = await cls.get_gen_table_by_id_service(auth, table_id) + if gen_table_info.id: try: # 确保options字段存在且为有效JSON if 'options' not in edit_gen_table or edit_gen_table['options'] is None: @@ -148,36 +172,42 @@ class GenTableService: except json.JSONDecodeError: edit_gen_table['options'] = '{}' - await gen_table_dao.update(id=page_object.table_id, data=edit_gen_table) - if page_object.columns: + await gen_table_dao.update(id=table_id, data=edit_gen_table) + if hasattr(page_object, 'columns') and page_object.columns: for gen_table_column in page_object.columns: - gen_table_column.update_by = page_object.update_by - gen_table_column.update_time = datetime.now() - if gen_table_column.column_id is not None: + # 为列添加更新信息 + gen_table_column_dict = gen_table_column.model_dump() + gen_table_column_dict['update_by'] = getattr(page_object, 'update_by', '') + gen_table_column_dict['update_time'] = datetime.now() + # 检查是否有id属性 + column_id = getattr(gen_table_column, 'id', None) + if column_id is not None: await gen_table_column_dao.update( - id=gen_table_column.column_id, - data=gen_table_column.model_dump(by_alias=True) + id=column_id, + data=gen_table_column_dict ) - await auth.db.commit() + if isinstance(db, AsyncSession): + await db.commit() return {"is_success": True, "message": "更新成功"} except Exception as e: - try: - await auth.db.rollback() - except: - pass # 忽略回滚错误 + if isinstance(db, AsyncSession): + try: + await db.rollback() + except: + pass # 忽略回滚错误 raise CustomException(msg=f'更新失败: {str(e)}') else: raise CustomException(msg='业务表不存在') @classmethod async def delete_gen_table_service(cls, auth: AuthSchema, page_object: GenTableDeleteSchema) -> SuccessResponse: - """ - 删除业务表信息service - - :param auth: 认证信息 - :param page_object: 删除业务表对象 - :return: 删除业务表校验结果 - """ + """删除业务表信息""" + if not auth.db: + raise CustomException(msg='数据库连接不存在') + # 确保db是AsyncSession类型 + db = auth.db + if not isinstance(db, AsyncSession): + raise CustomException(msg='数据库连接类型不正确') gen_table_dao = GenTableCRUD(auth=auth) gen_table_column_dao = GenTableColumnCRUD(auth=auth) @@ -188,32 +218,34 @@ class GenTableService: await gen_table_dao.delete(ids=[int(table_id)]) # 删除相关的字段信息 # 这里需要先查询出所有相关的column_id，然后删除 - columns = await gen_table_column_dao.get_gen_table_column_list_by_table_id(auth.db, int(table_id)) + columns = await gen_table_column_dao.get_gen_table_column_list_by_table_id(db, int(table_id)) if columns: column_ids = [column.id for column in columns] await gen_table_column_dao.delete(ids=column_ids) - await auth.db.commit() + if isinstance(db, AsyncSession): + await db.commit() return SuccessResponse(msg='删除成功') except Exception as e: - try: - await auth.db.rollback() - except: - pass # 忽略回滚错误 + if isinstance(db, AsyncSession): + try: + await db.rollback() + except: + pass # 忽略回滚错误 raise CustomException(msg=f'删除失败: {str(e)}') else: raise CustomException(msg='传入业务表id为空') @classmethod async def get_gen_table_by_id_service(cls, auth: AuthSchema, table_id: int) -> GenTableOutSchema: - """ - 获取需要生成的业务表详细信息service - - :param auth: 认证信息 - :param table_id: 需要生成的业务表id - :return: 需要生成的业务表id对应的信息 - """ + """获取需要生成的业务表详细信息""" + if not auth.db: + raise CustomException(msg='数据库连接不存在') + # 确保db是AsyncSession类型 + db = auth.db + if not isinstance(db, AsyncSession): + raise CustomException(msg='数据库连接类型不正确') gen_table_dao = GenTableCRUD(auth=auth) - gen_table = await gen_table_dao.get_gen_table_by_id(auth.db, table_id) + gen_table = await gen_table_dao.get_gen_table_by_id(db, table_id) if gen_table: result = await cls.set_table_from_options(GenTableOutSchema(**CamelCaseUtil.transform_result(gen_table))) return result @@ -221,15 +253,16 @@ class GenTableService: raise CustomException(msg='业务表不存在') @classmethod - async def get_gen_table_all_service(cls, auth: AuthSchema) -> list[GenTableOutSchema]: - """ - 获取所有业务表信息service - - :param auth: 认证信息 - :return: 所有业务表信息列表 - """ + async def get_gen_table_all_service(cls, auth: AuthSchema) -> List[GenTableOutSchema]: + """获取所有业务表信息""" + if not auth.db: + raise CustomException(msg='数据库连接不存在') + # 确保db是AsyncSession类型 + db = auth.db + if not isinstance(db, AsyncSession): + raise CustomException(msg='数据库连接类型不正确') gen_table_dao = GenTableCRUD(auth=auth) - gen_tables = await gen_table_dao.get_gen_table_all(auth.db) + gen_tables = await gen_table_dao.get_gen_table_all(db) result = [] for table in gen_tables: table_info = await cls.set_table_from_options(GenTableOutSchema(**CamelCaseUtil.transform_result(table))) @@ -237,38 +270,8 @@ class GenTableService: return result @classmethod - async def create_table_service(cls, auth: AuthSchema, sql: str, current_user: UserOutSchema) -> SuccessResponse: - """ - 创建表结构service - - :param auth: 认证信息 - :param sql: 建表语句 - :param current_user: 当前用户信息对象 - :return: 创建表结构结果 - """ - gen_table_dao = GenTableCRUD(auth=auth) - - try: - # 执行SQL语句创建表 - await gen_table_dao.create_table_by_sql_dao(auth.db, [sql]) - await auth.db.commit() - return SuccessResponse(msg='创建表结构成功') - except Exception as e: - try: - await auth.db.rollback() - except: - pass # 忽略回滚错误 - raise CustomException(msg=f'创建表结构失败: {str(e)}') - - @classmethod - async def preview_code_service(cls, auth: AuthSchema, table_id: int) -> dict[Any, Any]: - """ - 预览代码service - - :param auth: 认证信息 - :param table_id: 业务表id - :return: 预览数据列表 - """ + async def preview_code_service(cls, auth: AuthSchema, table_id: int) -> Dict[Any, Any]: + """预览代码""" gen_table = await cls.get_gen_table_by_id_service(auth, table_id) await cls.set_sub_table(auth, gen_table) await cls._set_pk_column(gen_table) @@ -286,13 +289,7 @@ class GenTableService: @classmethod async def generate_code_service(cls, auth: AuthSchema, table_name: str) -> SuccessResponse: - """ - 生成代码至指定路径service - - :param auth: 认证信息 - :param table_name: 业务表名称 - :return: 生成代码结果 - """ + """生成代码至指定路径""" env = TemplateInitializer.init_jinja2() render_info = await cls.__get_gen_render_info(auth, table_name) for template in render_info[0]: @@ -310,13 +307,7 @@ class GenTableService: @classmethod async def batch_gen_code_service(cls, auth: AuthSchema, table_names: List[str]) -> bytes: - """ - 批量生成代码service - - :param auth: 认证信息 - :param table_names: 业务表名称组 - :return: 下载代码结果 - """ + """批量生成代码""" zip_buffer = io.BytesIO() with zipfile.ZipFile(zip_buffer, 'w', zipfile.ZIP_DEFLATED) as zip_file: for table_name in table_names: @@ -330,71 +321,24 @@ class GenTableService: zip_buffer.close() return zip_data - @classmethod - async def __get_gen_render_info(cls, auth: AuthSchema, table_name: str) -> list[Any]: - """ - 获取生成代码渲染模板相关信息 - - :param auth: 认证信息 - :param table_name: 业务表名称 - :return: 生成代码渲染模板相关信息 - """ - - gen_table_dao = GenTableCRUD(auth=auth) - gen_table = await gen_table_dao.get_gen_table_by_name(auth.db, table_name) - if gen_table: - gen_table_schema = GenTableOutSchema(**CamelCaseUtil.transform_result(gen_table)) - await cls.set_sub_table(auth, gen_table_schema) - await cls._set_pk_column(gen_table_schema) - context = TemplateUtils.prepare_context(gen_table_schema) - template_list = TemplateUtils.get_template_list( - gen_table_schema.tpl_category or "", - gen_table_schema.tpl_web_type or "" - ) - output_files = [TemplateUtils.get_file_name([template], gen_table_schema)[0] for template in template_list] - - return [template_list, output_files, context, gen_table_schema] - else: - raise CustomException(msg=f'业务表 {table_name} 不存在') - - @classmethod - def __get_gen_path(cls, gen_table: GenTableOutSchema, template: str) -> Optional[str]: - """ - 根据GenTableModel对象和模板名称生成路径 - - :param gen_table: GenTableModel对象 - :param template: 模板名称 - :return: 生成的路径 - """ - try: - gen_path = gen_table.gen_path or "" - if gen_path == '/': - file_name = TemplateUtils.get_file_name([template], gen_table)[0] - return os.path.join(os.getcwd(), GEN_PATH, file_name) - else: - file_name = TemplateUtils.get_file_name([template], gen_table)[0] - return os.path.join(gen_path, file_name) - except Exception: - return None - @classmethod async def sync_db_service(cls, auth: AuthSchema, table_name: str) -> SuccessResponse: - """ - 同步数据库service - - :param auth: 认证信息 - :param table_name: 业务表名称 - :return: 同步数据库结果 - """ + """同步数据库""" + if not auth.db: + raise CustomException(msg='数据库连接不存在') + # 确保db是AsyncSession类型 + db = auth.db + if not isinstance(db, AsyncSession): + raise CustomException(msg='数据库连接类型不正确') gen_table_dao = GenTableCRUD(auth=auth) gen_table_column_dao = GenTableColumnCRUD(auth=auth) - gen_table = await gen_table_dao.get_gen_table_by_name(auth.db, table_name) + gen_table = await gen_table_dao.get_gen_table_by_name(db, table_name) if gen_table: table = GenTableOutSchema(**CamelCaseUtil.transform_result(gen_table)) table_columns = table.columns or [] # 确保不为None table_column_map = {column.column_name: column for column in table_columns} - query_db_table_columns = await gen_table_column_dao.get_gen_db_table_columns_by_name(auth.db, table_name) + query_db_table_columns = await gen_table_column_dao.get_gen_db_table_columns_by_name(db, table_name) db_table_columns = [ GenTableColumnOutSchema(**column) for column in CamelCaseUtil.transform_result(query_db_table_columns) ] @@ -406,7 +350,11 @@ class GenTableService: GenUtils.init_column_field(column, table) if column.column_name in table_column_map: prev_column = table_column_map[column.column_name] - column.column_id = prev_column.column_id + # 使用getattr安全访问id属性 + column_id = getattr(prev_column, 'id', None) + if column_id is not None: + # 为column设置id属性 + column.id = column_id if getattr(column, 'list', False): # 使用getattr安全访问属性 column.dict_type = prev_column.dict_type column.query_type = prev_column.query_type @@ -418,50 +366,50 @@ class GenTableService: ): column.is_required = prev_column.is_required column.html_type = prev_column.html_type - if column.column_id is not None: - await gen_table_column_dao.update(id=column.column_id, data=column.model_dump(by_alias=True)) + # 使用getattr安全访问id属性 + column_id = getattr(column, 'id', None) + if column_id is not None: + await gen_table_column_dao.update(id=column_id, data=column.model_dump(by_alias=True)) else: await gen_table_column_dao.create(data=column.model_dump(by_alias=True)) del_columns = [column for column in table_columns if column.column_name not in db_table_column_names] if del_columns: for column in del_columns: - if column.column_id is not None: - await gen_table_column_dao.delete(ids=[column.column_id]) - await auth.db.commit() + # 使用getattr安全访问id属性 + column_id = getattr(column, 'id', None) + if column_id is not None: + await gen_table_column_dao.delete(ids=[column_id]) + if isinstance(db, AsyncSession): + await db.commit() return SuccessResponse(msg='同步成功') except Exception as e: - try: - await auth.db.rollback() - except: - pass # 忽略回滚错误 + if isinstance(db, AsyncSession): + try: + await db.rollback() + except: + pass # 忽略回滚错误 raise CustomException(msg=f'同步失败: {str(e)}') else: raise CustomException('业务表不存在') @classmethod async def set_sub_table(cls, auth: AuthSchema, gen_table: GenTableOutSchema) -> None: - """ - 设置主子表信息 - - :param auth: 认证信息 - :param gen_table: 业务表信息 - :return: - """ - + """设置主子表信息""" + if not auth.db: + raise CustomException(msg='数据库连接不存在') + # 确保db是AsyncSession类型 + db = auth.db + if not isinstance(db, AsyncSession): + raise CustomException(msg='数据库连接类型不正确') if gen_table.sub_table_name: gen_table_dao = GenTableCRUD(auth=auth) - sub_table = await gen_table_dao.get_gen_table_by_name(auth.db, gen_table.sub_table_name) + sub_table = await gen_table_dao.get_gen_table_by_name(db, gen_table.sub_table_name) if sub_table: gen_table.sub_table = GenTableOutSchema(**CamelCaseUtil.transform_result(sub_table)) @classmethod async def _set_pk_column(cls, gen_table: GenTableOutSchema) -> None: - """ - 设置主键列信息 - - :param gen_table: 业务表信息 - :return: - """ + """设置主键列信息""" if gen_table.columns: for column in gen_table.columns: if column.pk: @@ -480,12 +428,7 @@ class GenTableService: @classmethod async def set_table_from_options(cls, gen_table: GenTableOutSchema) -> GenTableOutSchema: - """ - 设置代码生成其他选项值 - - :param gen_table: 生成对象 - :return: 设置后的生成对象 - """ + """设置代码生成其他选项值""" params_obj = json.loads(gen_table.options) if gen_table.options else None if params_obj: gen_table.tree_code = params_obj.get(GenConstant.TREE_CODE) @@ -497,12 +440,8 @@ class GenTableService: return gen_table @classmethod - async def validate_edit(cls, edit_gen_table: GenTableUpdateSchema): - """ - 编辑保存参数校验 - - :param edit_gen_table: 编辑业务表对象 - """ + async def validate_edit(cls, edit_gen_table: GenTableUpdateSchema) -> None: + """编辑保存参数校验""" if edit_gen_table.tpl_category == GenConstant.TPL_TREE: # 从options字段获取参数，而不是params if not edit_gen_table.options: @@ -522,24 +461,61 @@ class GenTableService: elif not edit_gen_table.sub_table_fk_name: raise CustomException(msg='子表关联的外键名不能为空') + @classmethod + async def __get_gen_render_info(cls, auth: AuthSchema, table_name: str) -> List[Any]: + """获取生成代码渲染模板相关信息""" + if not auth.db: + raise CustomException(msg='数据库连接不存在') + # 确保db是AsyncSession类型 + db = auth.db + if not isinstance(db, AsyncSession): + raise CustomException(msg='数据库连接类型不正确') + gen_table_dao = GenTableCRUD(auth=auth) + gen_table = await gen_table_dao.get_gen_table_by_name(db, table_name) + if gen_table: + gen_table_schema = GenTableOutSchema(**CamelCaseUtil.transform_result(gen_table)) + await cls.set_sub_table(auth, gen_table_schema) + await cls._set_pk_column(gen_table_schema) + context = TemplateUtils.prepare_context(gen_table_schema) + template_list = TemplateUtils.get_template_list( + gen_table_schema.tpl_category or "", + gen_table_schema.tpl_web_type or "" + ) + output_files = [TemplateUtils.get_file_name([template], gen_table_schema)[0] for template in template_list] + + return [template_list, output_files, context, gen_table_schema] + else: + raise CustomException(msg=f'业务表 {table_name} 不存在') + + @classmethod + def __get_gen_path(cls, gen_table: GenTableOutSchema, template: str) -> Optional[str]: + """根据GenTableModel对象和模板名称生成路径""" + try: + gen_path = gen_table.gen_path or "" + if gen_path == '/': + file_name = TemplateUtils.get_file_name([template], gen_table)[0] + return os.path.join(os.getcwd(), GEN_PATH, file_name) + else: + file_name = TemplateUtils.get_file_name([template], gen_table)[0] + return os.path.join(gen_path, file_name) + except Exception: + return None + class GenTableColumnService: - """ - 代码生成业务表字段服务层 - """ + """代码生成业务表字段服务层""" @classmethod async def get_gen_table_column_list_by_table_id_service(cls, auth: AuthSchema, table_id: int) -> List[GenTableColumnOutSchema]: - """ - 获取业务表字段列表信息service - - :param auth: 认证信息 - :param table_id: 业务表格id - :return: 业务表字段列表信息对象 - """ + """获取业务表字段列表信息""" + if not auth.db: + raise CustomException(msg='数据库连接不存在') + # 确保db是AsyncSession类型 + db = auth.db + if not isinstance(db, AsyncSession): + raise CustomException(msg='数据库连接类型不正确') gen_table_column_dao = GenTableColumnCRUD(auth=auth) - gen_table_column_list_result = await gen_table_column_dao.get_gen_table_column_list_by_table_id(auth.db, table_id) - + gen_table_column_list_result = await gen_table_column_dao.get_gen_table_column_list_by_table_id(db, table_id) return [ GenTableColumnOutSchema(**gen_table_column) for gen_table_column in CamelCaseUtil.transform_result(gen_table_column_list_result) diff --git a/backend/app/api/v1/module_monitor/job/crud.py b/backend/app/api/v1/module_monitor/job/crud.py index 3b263c35..ae2d6ad1 100644 --- a/backend/app/api/v1/module_monitor/job/crud.py +++ b/backend/app/api/v1/module_monitor/job/crud.py @@ -21,7 +21,7 @@ class JobCRUD(CRUDBase[JobModel, JobCreateSchema, JobUpdateSchema]): """获取定时任务详情""" return await self.get(id=id) - async def get_obj_list_crud(self, search: Dict = None, order_by: List[Dict[str, str]] = None) -> Sequence[JobModel]: + async def get_obj_list_crud(self, search: Optional[Dict] = None, order_by: Optional[List[Dict[str, str]]] = None) -> Sequence[JobModel]: """获取定时任务列表""" return await self.list(search=search, order_by=order_by) @@ -58,7 +58,7 @@ class JobLogCRUD(CRUDBase[JobLogModel, JobLogCreateSchema, JobLogUpdateSchema]): """获取定时任务日志详情""" return await self.get(id=id) - async def get_obj_log_list_crud(self, search: Dict = None, order_by: List[Dict[str, str]] = None) -> Sequence[JobLogModel]: + async def get_obj_log_list_crud(self, search: Optional[Dict] = None, order_by: Optional[List[Dict[str, str]]] = None) -> Sequence[JobLogModel]: """获取定时任务日志列表""" return await self.list(search=search, order_by=order_by) diff --git a/backend/app/api/v1/module_system/dept/crud.py b/backend/app/api/v1/module_system/dept/crud.py index dde282fd..720ae73f 100644 --- a/backend/app/api/v1/module_system/dept/crud.py +++ b/backend/app/api/v1/module_system/dept/crud.py @@ -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: """ diff --git a/backend/app/api/v1/module_system/dict/crud.py b/backend/app/api/v1/module_system/dict/crud.py index c5f0e268..e92c0ddd 100644 --- a/backend/app/api/v1/module_system/dict/crud.py +++ b/backend/app/api/v1/module_system/dict/crud.py @@ -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) diff --git a/backend/app/api/v1/module_system/log/crud.py b/backend/app/api/v1/module_system/log/crud.py index ff22bb1f..f88d3c7b 100644 --- a/backend/app/api/v1/module_system/log/crud.py +++ b/backend/app/api/v1/module_system/log/crud.py @@ -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]: """ 获取操作日志列表 diff --git a/backend/app/api/v1/module_system/menu/crud.py b/backend/app/api/v1/module_system/menu/crud.py index ba1816b7..7af763f1 100644 --- a/backend/app/api/v1/module_system/menu/crud.py +++ b/backend/app/api/v1/module_system/menu/crud.py @@ -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: """ diff --git a/backend/app/api/v1/module_system/menu/model.py b/backend/app/api/v1/module_system/menu/model.py index a7b1f30c..4b495459 100644 --- a/backend/app/api/v1/module_system/menu/model.py +++ b/backend/app/api/v1/module_system/menu/model.py @@ -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') diff --git a/backend/app/api/v1/module_system/notice/crud.py b/backend/app/api/v1/module_system/notice/crud.py index 4e07ab7b..aad22923 100644 --- a/backend/app/api/v1/module_system/notice/crud.py +++ b/backend/app/api/v1/module_system/notice/crud.py @@ -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) diff --git a/backend/app/api/v1/module_system/params/crud.py b/backend/app/api/v1/module_system/params/crud.py index 0a24df13..b68f9d5c 100644 --- a/backend/app/api/v1/module_system/params/crud.py +++ b/backend/app/api/v1/module_system/params/crud.py @@ -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) diff --git a/backend/app/api/v1/module_system/params/service.py b/backend/app/api/v1/module_system/params/service.py index 745739d8..19709493 100644 --- a/backend/app/api/v1/module_system/params/service.py +++ b/backend/app/api/v1/module_system/params/service.py @@ -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) diff --git a/backend/app/api/v1/module_system/position/controller.py b/backend/app/api/v1/module_system/position/controller.py index 52e25eea..b8d63c4c 100644 --- a/backend/app/api/v1/module_system/position/controller.py +++ b/backend/app/api/v1/module_system/position/controller.py @@ -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="查询岗位列表成功") diff --git a/backend/app/api/v1/module_system/position/crud.py b/backend/app/api/v1/module_system/position/crud.py index 79b09aa9..d7734c27 100644 --- a/backend/app/api/v1/module_system/position/crud.py +++ b/backend/app/api/v1/module_system/position/crud.py @@ -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]: """ 获取岗位列表 diff --git a/backend/app/api/v1/module_system/position/param.py b/backend/app/api/v1/module_system/position/param.py index 63d4fe03..35789c78 100644 --- a/backend/app/api/v1/module_system/position/param.py +++ b/backend/app/api/v1/module_system/position/param.py @@ -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)) diff --git a/backend/app/api/v1/module_system/position/service.py b/backend/app/api/v1/module_system/position/service.py index 26dccbfa..0fad68ba 100644 --- a/backend/app/api/v1/module_system/position/service.py +++ b/backend/app/api/v1/module_system/position/service.py @@ -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] diff --git a/backend/app/api/v1/module_system/role/controller.py b/backend/app/api/v1/module_system/role/controller.py index 2e0a0f7b..24271e61 100644 --- a/backend/app/api/v1/module_system/role/controller.py +++ b/backend/app/api/v1/module_system/role/controller.py @@ -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="查询角色成功") diff --git a/backend/app/api/v1/module_system/role/crud.py b/backend/app/api/v1/module_system/role/crud.py index cf5fe7d5..7b534a1d 100644 --- a/backend/app/api/v1/module_system/role/crud.py +++ b/backend/app/api/v1/module_system/role/crud.py @@ -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) diff --git a/backend/app/api/v1/module_system/role/service.py b/backend/app/api/v1/module_system/role/service.py index 014bf40d..43f80b55 100644 --- a/backend/app/api/v1/module_system/role/service.py +++ b/backend/app/api/v1/module_system/role/service.py @@ -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] diff --git a/backend/app/api/v1/module_system/user/controller.py b/backend/app/api/v1/module_system/user/controller.py index 5337cd83..e3622273 100644 --- a/backend/app/api/v1/module_system/user/controller.py +++ b/backend/app/api/v1/module_system/user/controller.py @@ -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: diff --git a/backend/app/api/v1/module_system/user/crud.py b/backend/app/api/v1/module_system/user/crud.py index 6936b0fd..7492935b 100644 --- a/backend/app/api/v1/module_system/user/crud.py +++ b/backend/app/api/v1/module_system/user/crud.py @@ -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]: """ 获取用户列表 diff --git a/backend/app/api/v1/module_system/user/param.py b/backend/app/api/v1/module_system/user/param.py index 611c47ce..c335e9ca 100644 --- a/backend/app/api/v1/module_system/user/param.py +++ b/backend/app/api/v1/module_system/user/param.py @@ -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)) diff --git a/backend/app/api/v1/module_system/user/service.py b/backend/app/api/v1/module_system/user/service.py index 3eb6620f..7ddf63ea 100644 --- a/backend/app/api/v1/module_system/user/service.py +++ b/backend/app/api/v1/module_system/user/service.py @@ -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: diff --git a/backend/app/common/request.py b/backend/app/common/request.py index ccbe96e4..087c095f 100644 --- a/backend/app/common/request.py +++ b/backend/app/common/request.py @@ -16,7 +16,7 @@ class PageResultSchema(BaseModel): page_size: Optional[int] = Field(default=None, ge=1, description="页面大小，默认为10") total: int = Field(default=0, ge=0, description="总记录数") has_next: Optional[bool] = Field(default=False, description="是否有下一页") - items: List[Any] = Field(default_factory=list, description="分页后的数据列表") + items: Optional[List[Any]] = Field(default_factory=list, description="分页后的数据列表") class PaginationService: diff --git a/backend/app/core/base_crud.py b/backend/app/core/base_crud.py index 4b65f6ab..8384cca5 100644 --- a/backend/app/core/base_crud.py +++ b/backend/app/core/base_crud.py @@ -1,20 +1,21 @@ # -*- coding: utf-8 -*- from pydantic import BaseModel -from typing import TypeVar, Sequence, Generic, Dict, Any, List, Union, Optional +from typing import TypeVar, Sequence, Generic, Dict, Any, List, Optional, Type from sqlalchemy.sql.elements import ColumnElement -from sqlalchemy.orm import Session, selectinload, DeclarativeBase +from sqlalchemy.orm import selectinload from sqlalchemy.engine import Result -from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy import asc, func, select, delete, Select, desc, update, or_, and_ +from app.core.base_model import MappedBase from app.api.v1.module_system.auth.schema import AuthSchema from app.api.v1.module_system.dept.model import DeptModel from app.api.v1.module_system.user.model import UserModel from app.utils.common_util import get_child_id_map, get_child_recursion from app.core.exceptions import CustomException +from app.common.request import PageResultSchema -ModelType = TypeVar("ModelType", bound=DeclarativeBase) +ModelType = TypeVar("ModelType", bound=MappedBase) CreateSchemaType = TypeVar("CreateSchemaType", bound=BaseModel) UpdateSchemaType = TypeVar("UpdateSchemaType", bound=BaseModel) @@ -22,17 +23,17 @@ UpdateSchemaType = TypeVar("UpdateSchemaType", bound=BaseModel) class CRUDBase(Generic[ModelType, CreateSchemaType, UpdateSchemaType]): """基础数据层""" - def __init__(self, model: ModelType, auth: AuthSchema) -> None: + def __init__(self, model: Type[ModelType], auth: AuthSchema) -> None: """ 初始化CRUDBase类 Args: - model: 数据模型 + model: 数据模型类 auth: 认证信息 """ self.model = model self.auth = auth - self.db: AsyncSession | Session | None = auth.db + self.db = auth.db self.current_user = auth.user async def get(self, **kwargs) -> Optional[ModelType]: @@ -50,8 +51,7 @@ class CRUDBase(Generic[ModelType, CreateSchemaType, UpdateSchemaType]): """ try: conditions = await self.__build_conditions(**kwargs) - sql = (select(self.model) - .where(*conditions)) + sql = select(self.model).where(*conditions) # 只有继承自CreatorMixin的模型才有creator关系 if hasattr(self.model, "creator_id"): sql = sql.options(selectinload(self.model.creator)) @@ -67,7 +67,7 @@ class CRUDBase(Generic[ModelType, CreateSchemaType, UpdateSchemaType]): except Exception as e: raise CustomException(msg=f"获取查询失败: {str(e)}") - async def list(self, search: Dict = None, order_by: List[Dict[str, str]] = None) -> Sequence[ModelType]: + async def list(self, search: Optional[Dict] = None, order_by: Optional[List[Dict[str, str]]] = None) -> Sequence[ModelType]: """ 根据条件获取对象列表和总数 @@ -84,9 +84,7 @@ class CRUDBase(Generic[ModelType, CreateSchemaType, UpdateSchemaType]): try: conditions = await self.__build_conditions(**search) if search else [] order = order_by or [{'id': 'asc'}] - sql = (select(self.model) - .where(*conditions) - .order_by(*self.__order_by(order))) + sql = select(self.model).where(*conditions).order_by(*self.__order_by(order)) # 只有继承自CreatorMixin的模型才有creator关系 if hasattr(self.model, "creator_id"): sql = sql.options(selectinload(self.model.creator)) @@ -96,7 +94,84 @@ class CRUDBase(Generic[ModelType, CreateSchemaType, UpdateSchemaType]): except Exception as e: raise CustomException(msg=f"列表查询失败: {str(e)}") - async def create(self, data: Union[CreateSchemaType, Dict]) -> ModelType: + async def tree_list(self, search: Optional[Dict] = None, order_by: Optional[List[Dict[str, str]]] = None, children_attr: str = 'children') -> Sequence[ModelType]: + """ + 获取树形结构数据列表 + + Args: + search: 查询条件 + order_by: 排序字段 + children_attr: 子节点属性名 + + Returns: + Sequence[ModelType]: 树形结构数据列表 + + Raises: + CustomException: 查询失败时抛出异常 + """ + try: + from sqlalchemy.orm import selectinload + + conditions = await self.__build_conditions(**search) if search else [] + order = order_by or [{'id': 'asc'}] + sql = select(self.model).where(*conditions).order_by(*self.__order_by(order)) + + # 如果模型有children属性，则预加载该关系 + if hasattr(self.model, children_attr): + sql = sql.options(selectinload(getattr(self.model, children_attr))) + + # 只有继承自CreatorMixin的模型才有creator关系 + if hasattr(self.model, "creator_id"): + sql = sql.options(selectinload(self.model.creator)) + + sql = await self.__filter_permissions(sql) + result: Result = await self.db.execute(sql) + return result.scalars().all() + except Exception as e: + raise CustomException(msg=f"树形列表查询失败: {str(e)}") + + async def page(self, offset: int, limit: int, order_by: List[Dict[str, str]], search: Dict) -> Dict: + try: + from sqlalchemy.orm import selectinload + + conditions = await self.__build_conditions(**search) if search else [] + order = order_by or [{'id': 'asc'}] + sql = select(self.model).where(*conditions).order_by(*self.__order_by(order)) + # 只有继承自CreatorMixin的模型才有creator关系 + if hasattr(self.model, "creator_id"): + sql = sql.options(selectinload(self.model.creator)) + sql = await self.__filter_permissions(sql) + + # 获取总数 + count_sql = select(func.count()).select_from(self.model) + # 应用相同的过滤条件到计数查询 + if conditions: + count_sql = count_sql.where(*conditions) + count_sql = await self.__filter_permissions(count_sql) + + total_result = await self.db.execute(count_sql) + total = total_result.scalar() + + if total is None: + total = 0 + + result: Result = await self.db.execute(sql.offset(offset).limit(limit)) + + objs = result.scalars().all() + + data=PageResultSchema( + items=objs, + total=total, + page_no=offset // limit + 1 if limit else 1, + page_size=limit, + has_next=offset + limit < total, + ).model_dump() + + return data + except Exception as e: + raise CustomException(msg=f"分页查询失败: {str(e)}") + + async def create(self, data: CreateSchemaType) -> ModelType: """ 创建新对象 @@ -127,7 +202,7 @@ class CRUDBase(Generic[ModelType, CreateSchemaType, UpdateSchemaType]): except Exception as e: raise CustomException(msg=f"创建失败: {str(e)}") - async def update(self, id: int, data: Union[UpdateSchemaType, Dict]) -> ModelType: + async def update(self, id: int, data: UpdateSchemaType) -> ModelType: """ 更新对象 @@ -226,7 +301,7 @@ class CRUDBase(Generic[ModelType, CreateSchemaType, UpdateSchemaType]): except Exception as e: raise CustomException(msg=f"更新关系失败: {str(e)}") - async def __filter_permissions(self, sql: Select[Any]) -> Select[Any]: + async def __filter_permissions(self, sql: Select) -> Select: """过滤数据权限""" # 如果不需要检查数据权限,则直接返回 if not self.current_user or not self.auth.check_data_scope: @@ -352,42 +427,4 @@ class CRUDBase(Generic[ModelType, CreateSchemaType, UpdateSchemaType]): conditions.append(getattr(attr, seq.replace("==", "__eq__"))(val)) else: conditions.append(attr == value) - return conditions - - async def get_tree_list(self, search: Dict = None, order_by: List[Dict[str, str]] = None, children_attr: str = 'children') -> Sequence[ModelType]: - """ - 获取树形结构数据列表 - - Args: - search: 查询条件 - order_by: 排序字段 - children_attr: 子节点属性名 - - Returns: - Sequence[ModelType]: 树形结构数据列表 - - Raises: - CustomException: 查询失败时抛出异常 - """ - try: - from sqlalchemy.orm import selectinload - - conditions = await self.__build_conditions(**search) if search else [] - order = order_by or [{'id': 'asc'}] - sql = (select(self.model) - .where(*conditions) - .order_by(*self.__order_by(order))) - - # 如果模型有children属性，则预加载该关系 - if hasattr(self.model, children_attr): - sql = sql.options(selectinload(getattr(self.model, children_attr))) - - # 只有继承自CreatorMixin的模型才有creator关系 - if hasattr(self.model, "creator_id"): - sql = sql.options(selectinload(self.model.creator)) - - sql = await self.__filter_permissions(sql) - result: Result = await self.db.execute(sql) - return result.scalars().all() - except Exception as e: - raise CustomException(msg=f"树形列表查询失败: {str(e)}") \ No newline at end of file + return conditions \ No newline at end of file diff --git a/backend/app/core/base_params.py b/backend/app/core/base_params.py index 8c2cb74f..3c5a7361 100644 --- a/backend/app/core/base_params.py +++ b/backend/app/core/base_params.py @@ -1,6 +1,6 @@ # -*- coding: utf-8 -*- -from typing import Optional +from typing import Optional, List, Dict from fastapi import Query @@ -11,7 +11,7 @@ class PaginationQueryParam: self, page_no: Optional[int] = Query(default=None, description="当前页码", ge=1), page_size: Optional[int] = Query(default=None, description="每页数量", ge=1, le=100), - order_by: Optional[str] = Query(default=None, description="排序字段,格式:[{'field':'asc/desc'}]"), + order_by: Optional[List[Dict[str, str]]] = Query(default=None, description="排序字段,格式:[{'field':'asc/desc'}]"), ) -> None: """ 初始化分页查询参数 diff --git a/backend/app/core/serialize.py b/backend/app/core/serialize.py new file mode 100644 index 00000000..dfeeff7a --- /dev/null +++ b/backend/app/core/serialize.py @@ -0,0 +1,71 @@ +# -*- coding: utf-8 -*- + +from pydantic import BaseModel +from typing import TypeVar, Dict, Any, Type, Generic +from sqlalchemy.orm import DeclarativeBase + +ModelType = TypeVar("ModelType", bound=DeclarativeBase) +SchemaType = TypeVar("SchemaType", bound=BaseModel) + + +class Serialize(Generic[ModelType, SchemaType]): + """ + 序列化工具类，提供模型、Schema和字典之间的转换功能 + """ + + @staticmethod + def schema_to_model(schema: SchemaType, model: Type[ModelType]) -> ModelType: + """ + 将Pydantic Schema转换为SQLAlchemy模型 + + Args: + schema: Pydantic Schema实例 + model: SQLAlchemy模型类 + + Returns: + SQLAlchemy模型实例 + + Raises: + Exception: 转换过程中可能抛出的异常 + """ + try: + return model(**schema.model_dump()) + except Exception as e: + raise ValueError(f"序列化失败: {str(e)}") + + @staticmethod + def model_to_schema(model: ModelType, schema: Type[SchemaType]) -> SchemaType: + """ + 将SQLAlchemy模型转换为Pydantic Schema + + Args: + model: SQLAlchemy模型实例 + schema: Pydantic Schema类 + + Returns: + Pydantic Schema实例 + + Raises: + Exception: 转换过程中可能抛出的异常 + """ + try: + return schema.model_validate(model) + except Exception as e: + raise ValueError(f"反序列化失败: {str(e)}") + + @staticmethod + def schema_to_dict( + schema: SchemaType, + ) -> Dict[str, Any]: + """ + 将Pydantic Schema转换为字典 + + Args: + schema: Pydantic Schema实例 + include: 包含的字段 + exclude: 排除的字段 + + Returns: + 包含Schema数据的字典 + """ + return schema.model_dump() \ No newline at end of file