mirror of
https://github.com/fastapiadmin/FastapiAdmin.git
synced 2026-09-22 13:05:18 +00:00
zhangtao
60fe97afc3
- 重构GenTableDao和GenTableColumnDao，继承CRUDBase，添加权限鉴权支持 - 优化分页逻辑，支持非分页时返回完整结果集 - 调整查询条件，支持大小写不敏感模糊查询和时间范围过滤 - 统一接口响应格式，返回SuccessResponse或ErrorResponse - 替换旧的依赖和注解，改用新的认证和日志中间件 - 文档README.en.md和README.md内容格式及示例完善与优化 - alembic/env.py中数据库URL配置优化，增加异常检测保障环境配置正确 - 代码生成控制器genController新增代码批量生成下载流和本地生成文件覆盖检测 - 删除或更新废弃代码，清理无用导入，提升项目整体代码质量和可维护性
307 lines
9.4 KiB
Python
307 lines
9.4 KiB
Python
# -*- coding: utf-8 -*-
|
|
|
|
from pathlib import Path
|
|
|
|
|
|
import importlib
|
|
import os
|
|
import uuid
|
|
import re
|
|
|
|
from sqlalchemy.orm import DeclarativeBase
|
|
from typing import Any, Generator, List, Dict, Sequence, Optional
|
|
|
|
from app.config import setting
|
|
from app.core.logger import logger
|
|
from app.core.exceptions import CustomException
|
|
|
|
def worship():
|
|
print("""
|
|
______ _ _
|
|
| ____| | | /\ (_)
|
|
| |__ __ _ ___| |_ / \ _ __ _
|
|
| __/ _` / __| __| / /\ \ | '_ \| |
|
|
| | | (_| \__ \ |_ / ____ \| |_) | |
|
|
|_| \__,_|___/\__/_/ \_\ .__/|_|
|
|
| |
|
|
|_|
|
|
""")
|
|
|
|
def import_module(module: str, desc: str) -> Any:
|
|
"""
|
|
动态导入模块
|
|
:param module: 模块名称
|
|
:param desc: 模块描述
|
|
:return: 模块对象
|
|
"""
|
|
try:
|
|
module_path, module_class = module.rsplit(".", 1)
|
|
module = importlib.import_module(module_path)
|
|
return getattr(module, module_class)
|
|
except ModuleNotFoundError:
|
|
logger.error(f"导入{desc}失败,未找到模块:{module}")
|
|
raise ModuleNotFoundError(f"导入{desc}失败,未找到模块:{module}")
|
|
except AttributeError:
|
|
logger.error(f"导入{desc}失败,未找到模块方法:{module}")
|
|
raise AttributeError(f"导入{desc}失败,未找到模块方法:{module}")
|
|
|
|
|
|
async def import_modules_async(modules: list, desc: str, **kwargs):
|
|
"""
|
|
异步导入模块列表
|
|
:param modules: 模块列表
|
|
:param desc: 模块描述
|
|
:param kwargs: 额外参数
|
|
"""
|
|
for module in modules:
|
|
if not module:
|
|
continue
|
|
try:
|
|
module_path = module[0:module.rindex(".")]
|
|
module_name = module[module.rindex(".") + 1:]
|
|
module_obj = importlib.import_module(module_path)
|
|
await getattr(module_obj, module_name)(**kwargs)
|
|
except ModuleNotFoundError:
|
|
logger.error(f"导入{desc}失败,未找到模块:{module}")
|
|
raise ModuleNotFoundError(f"导入{desc}失败,未找到模块:{module}")
|
|
except AttributeError:
|
|
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映射字典
|
|
:param model_list: 模型列表
|
|
:return: {id: parent_id} 映射字典
|
|
"""
|
|
return {item.id: item.parent_id for item in model_list}
|
|
|
|
|
|
def get_parent_recursion(
|
|
id: int,
|
|
id_map: Dict[int, int],
|
|
ids: Optional[List[int]] = None
|
|
) -> List[int]:
|
|
"""
|
|
递归获取所有父级ID
|
|
:param id: 当前ID
|
|
:param id_map: ID映射字典
|
|
:param ids: 已收集的ID列表
|
|
:return: 所有父级ID列表
|
|
"""
|
|
ids = ids or []
|
|
if id in ids:
|
|
raise CustomException(msg="递归获取父级ID失败,不可以自引用")
|
|
ids.append(id)
|
|
parent_id = id_map.get(id)
|
|
if parent_id:
|
|
get_parent_recursion(parent_id, id_map, ids)
|
|
return ids
|
|
|
|
|
|
def get_child_id_map(model_list: Sequence[DeclarativeBase]) -> Dict[int, List[int]]:
|
|
"""
|
|
获取子级ID映射字典
|
|
:param model_list: 模型列表
|
|
:return: {id: [child_ids]} 映射字典
|
|
"""
|
|
data_map = {}
|
|
for model in model_list:
|
|
data_map.setdefault(model.id, [])
|
|
if model.parent_id:
|
|
data_map.setdefault(model.parent_id, []).append(model.id)
|
|
return data_map
|
|
|
|
|
|
def get_child_recursion(
|
|
id: int,
|
|
id_map: Dict[int, List[int]],
|
|
ids: Optional[List[int]] = None
|
|
) -> List[int]:
|
|
"""
|
|
递归获取所有子级ID
|
|
:param id: 当前ID
|
|
:param id_map: ID映射字典
|
|
:param ids: 已收集的ID列表
|
|
:return: 所有子级ID列表
|
|
"""
|
|
ids = ids or []
|
|
ids.append(id)
|
|
for child in id_map.get(id, []):
|
|
get_child_recursion(child, id_map, ids)
|
|
return ids
|
|
|
|
|
|
def traversal_to_tree(nodes: list[dict[str, Any]]) -> list[dict[str, Any]]:
|
|
"""
|
|
通过遍历算法构造树形结构
|
|
|
|
:param nodes: 树节点列表
|
|
:return:
|
|
"""
|
|
tree: list[dict[str, Any]] = []
|
|
node_dict = {node['id']: node for node in nodes}
|
|
|
|
for node in nodes:
|
|
# 确保每个节点都有children字段，即使没有子节点也设置为null
|
|
if 'children' not in node:
|
|
node['children'] = None
|
|
|
|
parent_id = node['parent_id']
|
|
if parent_id is None:
|
|
tree.append(node)
|
|
else:
|
|
parent_node = node_dict.get(parent_id)
|
|
if parent_node is not None:
|
|
if 'children' not in parent_node or parent_node['children'] is None:
|
|
parent_node['children'] = []
|
|
if node not in parent_node['children']:
|
|
parent_node['children'].append(node)
|
|
else:
|
|
if node not in tree:
|
|
tree.append(node)
|
|
|
|
# 确保所有节点都有children字段
|
|
for node in tree:
|
|
if 'children' not in node:
|
|
node['children'] = None
|
|
|
|
return tree
|
|
|
|
|
|
def recursive_to_tree(nodes: list[dict[str, Any]], *, parent_id: int | None = None) -> list[dict[str, Any]]:
|
|
"""
|
|
通过递归算法构造树形结构（性能影响较大）
|
|
|
|
:param nodes: 树节点列表
|
|
:param parent_id: 父节点 ID，默认为 None 表示根节点
|
|
:return:
|
|
"""
|
|
tree: list[dict[str, Any]] = []
|
|
for node in nodes:
|
|
if node['parent_id'] == parent_id:
|
|
child_nodes = recursive_to_tree(nodes, parent_id=node['id'])
|
|
if child_nodes:
|
|
node['children'] = child_nodes
|
|
tree.append(node)
|
|
return tree
|
|
|
|
|
|
def bytes2human(n: int, format_str: str = '%(value).1f%(symbol)s') -> str:
|
|
"""
|
|
字节数转人类可读格式
|
|
:param n: 字节数
|
|
:param format_str: 格式化字符串
|
|
:return: 可读的字节字符串,如 '1.5MB'
|
|
"""
|
|
symbols = ('B', 'KB', 'MB', 'GB', 'TB', 'PB', 'EB', 'ZB', 'YB')
|
|
prefix = {s: 1 << (i + 1) * 10 for i, s in enumerate(symbols[1:])}
|
|
for symbol in reversed(symbols[1:]):
|
|
if n >= prefix[symbol]:
|
|
value = float(n) / prefix[symbol]
|
|
return format_str % locals()
|
|
return format_str % dict(symbol=symbols[0], value=n)
|
|
|
|
|
|
def bytes2file_response(bytes_info: bytes) -> Generator[bytes, Any, None]:
|
|
"""生成文件响应"""
|
|
yield bytes_info
|
|
|
|
|
|
def get_filepath_from_url(url: str) -> Path:
|
|
"""
|
|
工具方法：根据请求参数获取文件路径
|
|
|
|
:param url: 请求参数中的url参数
|
|
:return: 文件路径
|
|
"""
|
|
file_info = url.split('?')[1].split('&')
|
|
task_id = file_info[0].split('=')[1]
|
|
file_name = file_info[1].split('=')[1]
|
|
task_path = file_info[2].split('=')[1]
|
|
filepath = setting.settings.STATIC_ROOT.joinpath(task_path, task_id, file_name)
|
|
|
|
return filepath
|
|
|
|
|
|
def export_list2excel(list_data: List) -> Any:
|
|
"""
|
|
工具方法：将需要导出的list数据转化为对应excel的二进制数据
|
|
|
|
:param list_data: 数据列表
|
|
:return: 字典信息对应excel的二进制数据
|
|
"""
|
|
df = pd.DataFrame(list_data)
|
|
binary_data = io.BytesIO()
|
|
df.to_excel(binary_data, index=False, engine='openpyxl')
|
|
binary_data = binary_data.getvalue()
|
|
|
|
return binary_data
|
|
|
|
|
|
def get_excel_template(header_list: List, selector_header_list: List, option_list: List[dict]) -> Any:
|
|
"""
|
|
工具方法：将需要导出的list数据转化为对应excel的二进制数据
|
|
|
|
:param header_list: 表头数据列表
|
|
:param selector_header_list: 需要设置为选择器格式的表头数据列表
|
|
:param option_list: 选择器格式的表头预设的选项列表
|
|
:return: 模板excel的二进制数据
|
|
"""
|
|
# 创建Excel工作簿
|
|
wb = Workbook()
|
|
# 选择默认的活动工作表
|
|
ws = wb.active
|
|
|
|
# 设置表头文字
|
|
headers = header_list
|
|
|
|
# 设置表头背景样式为灰色，前景色为白色
|
|
header_fill = PatternFill(start_color='ababab', end_color='ababab', fill_type='solid')
|
|
|
|
# 将表头写入第一行
|
|
for col_num, header in enumerate(headers, 1):
|
|
cell = ws.cell(row=1, column=col_num)
|
|
cell.value = header
|
|
cell.fill = header_fill
|
|
# 设置列宽度为16
|
|
ws.column_dimensions[chr(64 + col_num)].width = 12
|
|
# 设置水平居中对齐
|
|
cell.alignment = Alignment(horizontal='center')
|
|
|
|
# 设置选择器的预设选项
|
|
options = option_list
|
|
|
|
# 获取selector_header的字母索引
|
|
for selector_header in selector_header_list:
|
|
column_selector_header_index = headers.index(selector_header) + 1
|
|
|
|
# 创建数据有效性规则
|
|
header_option = []
|
|
for option in options:
|
|
if option.get(selector_header):
|
|
header_option = option.get(selector_header)
|
|
dv = DataValidation(type='list', formula1=f'"{",".join(header_option)}"')
|
|
# 设置数据有效性规则的起始单元格和结束单元格
|
|
dv.add(
|
|
f'{get_column_letter(column_selector_header_index)}2:{get_column_letter(column_selector_header_index)}1048576'
|
|
)
|
|
# 添加数据有效性规则到工作表
|
|
ws.add_data_validation(dv)
|
|
|
|
# 保存Excel文件为字节类型的数据
|
|
file = io.BytesIO()
|
|
wb.save(file)
|
|
file.seek(0)
|
|
|
|
# 读取字节数据
|
|
excel_data = file.getvalue()
|
|
|
|
return excel_data
|