refactor(gen): 重构代码生成模块,实现权限认证和分页支持

- 重构GenTableDao和GenTableColumnDao,继承CRUDBase,添加权限鉴权支持
- 优化分页逻辑,支持非分页时返回完整结果集
- 调整查询条件,支持大小写不敏感模糊查询和时间范围过滤
- 统一接口响应格式,返回SuccessResponse或ErrorResponse
- 替换旧的依赖和注解,改用新的认证和日志中间件
- 文档README.en.md和README.md内容格式及示例完善与优化
- alembic/env.py中数据库URL配置优化,增加异常检测保障环境配置正确
- 代码生成控制器genController新增代码批量生成下载流和本地生成文件覆盖检测
- 删除或更新废弃代码,清理无用导入,提升项目整体代码质量和可维护性
This commit is contained in:
zhangtao
2025-09-20 01:52:23 +08:00
parent 81d131c0a4
commit 60fe97afc3
14 changed files with 649 additions and 888 deletions
+14 -124
View File
@@ -1,12 +1,15 @@
# -*- coding: utf-8 -*-
from pathlib import Path
import importlib
import os
import uuid
import re
from sqlalchemy.orm import DeclarativeBase
from typing import Any, List, Dict, Sequence, Optional
from typing import Any, Generator, List, Dict, Sequence, Optional
from app.config import setting
from app.core.logger import logger
@@ -42,6 +45,7 @@ def import_module(module: str, desc: str) -> Any:
logger.error(f"导入{desc}失败,未找到模块方法:{module}")
raise AttributeError(f"导入{desc}失败,未找到模块方法:{module}")
async def import_modules_async(modules: list, desc: str, **kwargs):
"""
异步导入模块列表
@@ -64,10 +68,12 @@ async def import_modules_async(modules: list, desc: str, **kwargs):
logger.error(f"导入{desc}失败,未找到模块方法:{module}")
raise AttributeError(f"导入{desc}失败,未找到模块方法:{module}")
def get_random_character() -> str:
"""生成随机字符串"""
return uuid.uuid4().hex
def get_parent_id_map(model_list: Sequence[DeclarativeBase]) -> Dict[int, int]:
"""
获取父级ID映射字典
@@ -76,6 +82,7 @@ def get_parent_id_map(model_list: Sequence[DeclarativeBase]) -> Dict[int, int]:
"""
return {item.id: item.parent_id for item in model_list}
def get_parent_recursion(
id: int,
id_map: Dict[int, int],
@@ -97,6 +104,7 @@ def get_parent_recursion(
get_parent_recursion(parent_id, id_map, ids)
return ids
def get_child_id_map(model_list: Sequence[DeclarativeBase]) -> Dict[int, List[int]]:
"""
获取子级ID映射字典
@@ -110,6 +118,7 @@ def get_child_id_map(model_list: Sequence[DeclarativeBase]) -> Dict[int, List[in
data_map.setdefault(model.parent_id, []).append(model.id)
return data_map
def get_child_recursion(
id: int,
id_map: Dict[int, List[int]],
@@ -200,12 +209,12 @@ def bytes2human(n: int, format_str: str = '%(value).1f%(symbol)s') -> str:
return format_str % dict(symbol=symbols[0], value=n)
def bytes2file_response(bytes_info: bytes):
def bytes2file_response(bytes_info: bytes) -> Generator[bytes, Any, None]:
"""生成文件响应"""
yield bytes_info
def get_filepath_from_url(url: str):
def get_filepath_from_url(url: str) -> Path:
"""
工具方法:根据请求参数获取文件路径
@@ -221,126 +230,7 @@ def get_filepath_from_url(url: str):
return filepath
class SqlalchemyUtil:
"""
sqlalchemy工具类
"""
@classmethod
def base_to_dict(
cls, obj: Union[Base, Dict], transform_case: Literal['no_case', 'snake_to_camel', 'camel_to_snake'] = 'no_case'
):
"""
将sqlalchemy模型对象转换为字典
:param obj: sqlalchemy模型对象或普通字典
:param transform_case: 转换得到的结果形式,可选的有'no_case'(不转换)、'snake_to_camel'(下划线转小驼峰)、'camel_to_snake'(小驼峰转下划线),默认为'no_case'
:return: 字典结果
"""
if isinstance(obj, Base):
base_dict = obj.__dict__.copy()
base_dict.pop('_sa_instance_state', None)
for name, value in base_dict.items():
if isinstance(value, InstrumentedList):
base_dict[name] = cls.serialize_result(value, 'snake_to_camel')
elif isinstance(obj, dict):
base_dict = obj.copy()
if transform_case == 'snake_to_camel':
return {CamelCaseUtil.snake_to_camel(k): v for k, v in base_dict.items()}
elif transform_case == 'camel_to_snake':
return {SnakeCaseUtil.camel_to_snake(k): v for k, v in base_dict.items()}
return base_dict
@classmethod
def serialize_result(
cls, result: Any, transform_case: Literal['no_case', 'snake_to_camel', 'camel_to_snake'] = 'no_case'
):
"""
将sqlalchemy查询结果序列化
:param result: sqlalchemy查询结果
:param transform_case: 转换得到的结果形式,可选的有'no_case'(不转换)、'snake_to_camel'(下划线转小驼峰)、'camel_to_snake'(小驼峰转下划线),默认为'no_case'
:return: 序列化结果
"""
if isinstance(result, (Base, dict)):
return cls.base_to_dict(result, transform_case)
elif isinstance(result, list):
return [cls.serialize_result(row, transform_case) for row in result]
elif isinstance(result, Row):
if all([isinstance(row, Base) for row in result]):
return [cls.base_to_dict(row, transform_case) for row in result]
elif any([isinstance(row, Base) for row in result]):
return [cls.serialize_result(row, transform_case) for row in result]
else:
result_dict = result._asdict()
if transform_case == 'snake_to_camel':
return {CamelCaseUtil.snake_to_camel(k): v for k, v in result_dict.items()}
elif transform_case == 'camel_to_snake':
return {SnakeCaseUtil.camel_to_snake(k): v for k, v in result_dict.items()}
return result_dict
return result
class CamelCaseUtil:
"""
下划线形式(snake_case)转小驼峰形式(camelCase)工具方法
"""
@classmethod
def snake_to_camel(cls, snake_str: str):
"""
下划线形式字符串(snake_case)转换为小驼峰形式字符串(camelCase)
:param snake_str: 下划线形式字符串
:return: 小驼峰形式字符串
"""
# 分割字符串
words = snake_str.split('_')
# 小驼峰命名,第一个词首字母小写,其余词首字母大写
return words[0] + ''.join(word.capitalize() for word in words[1:])
@classmethod
def transform_result(cls, result: Any):
"""
针对不同类型将下划线形式(snake_case)批量转换为小驼峰形式(camelCase)方法
:param result: 输入数据
:return: 小驼峰形式结果
"""
return SqlalchemyUtil.serialize_result(result=result, transform_case='snake_to_camel')
class SnakeCaseUtil:
"""
小驼峰形式(camelCase)转下划线形式(snake_case)工具方法
"""
@classmethod
def camel_to_snake(cls, camel_str: str):
"""
小驼峰形式字符串(camelCase)转换为下划线形式字符串(snake_case)
:param camel_str: 小驼峰形式字符串
:return: 下划线形式字符串
"""
# 在大写字母前添加一个下划线,然后将整个字符串转为小写
words = re.sub('(.)([A-Z][a-z]+)', r'\1_\2', camel_str)
return re.sub('([a-z0-9])([A-Z])', r'\1_\2', words).lower()
@classmethod
def transform_result(cls, result: Any):
"""
针对不同类型将下划线形式(snake_case)批量转换为小驼峰形式(camelCase)方法
:param result: 输入数据
:return: 小驼峰形式结果
"""
return SqlalchemyUtil.serialize_result(result=result, transform_case='camel_to_snake')
def export_list2excel(list_data: List):
def export_list2excel(list_data: List) -> Any:
"""
工具方法:将需要导出的list数据转化为对应excel的二进制数据
@@ -355,7 +245,7 @@ def export_list2excel(list_data: List):
return binary_data
def get_excel_template(header_list: List, selector_header_list: List, option_list: List[dict]):
def get_excel_template(header_list: List, selector_header_list: List, option_list: List[dict]) -> Any:
"""
工具方法:将需要导出的list数据转化为对应excel的二进制数据