refactor(backend): 重构项目结构,将模块代码移动到plugin目录

feat(backend): 添加数据库表创建和删除工具函数
fix(backend): 修复日志输出和Redis连接提示信息
style(frontend): 调整导航栏头像大小
perf(backend): 优化系统配置和字典数据缓存逻辑
chore(backend): 更新模板文件路径配置
This commit is contained in:
zhangtao
2026-01-12 01:14:06 +08:00
parent 4bae515bfe
commit 646cbd6e60
68 changed files with 180 additions and 584 deletions
@@ -0,0 +1,2 @@
# -*- coding: utf-8 -*-
@@ -0,0 +1,253 @@
# -*- coding: utf-8 -*-
from typing import List
from fastapi import APIRouter, Depends, Body, Path
from fastapi.responses import JSONResponse
from app.common.response import SuccessResponse, StreamResponse
from app.core.dependencies import AuthPermission
from app.core.base_params import PaginationQueryParam
from app.common.request import PaginationService
from app.core.router_class import OperationLogRoute
from app.utils.common_util import bytes2file_response
from app.core.logger import log
from app.api.v1.module_system.auth.schema import AuthSchema
from .schema import GenTableSchema, GenTableQueryParam
from .service import GenTableService
GenRouter = APIRouter(route_class=OperationLogRoute, prefix='/gencode', tags=["代码生成模块"])
@GenRouter.get("/list", summary="查询代码生成业务表列表", description="查询代码生成业务表列表")
async def gen_table_list_controller(
page: PaginationQueryParam = Depends(),
search: GenTableQueryParam = Depends(),
auth: AuthSchema = Depends(AuthPermission(["module_generator:gencode:query"]))
) -> JSONResponse:
"""
查询代码生成业务表列表
参数:
- page (PaginationQueryParam): 分页查询参数
- search (GenTableQueryParam): 搜索参数
- auth (AuthSchema): 认证信息模型
返回:
- JSONResponse: 包含查询结果和分页信息的JSON响应
"""
result_dict_list = await GenTableService.get_gen_table_list_service(auth=auth, search=search)
result_dict = await PaginationService.paginate(data_list=result_dict_list, page_no=page.page_no, page_size=page.page_size)
log.info('获取代码生成业务表列表成功')
return SuccessResponse(data=result_dict, msg="获取代码生成业务表列表成功")
@GenRouter.get("/db/list", summary="查询数据库表列表", description="查询数据库表列表")
async def get_gen_db_table_list_controller(
page: PaginationQueryParam = Depends(),
search: GenTableQueryParam = Depends(),
auth: AuthSchema = Depends(AuthPermission(["module_generator:dblist:query"]))
) -> JSONResponse:
"""
查询数据库表列表
参数:
- page (PaginationQueryParam): 分页查询参数
- search (GenTableQueryParam): 搜索参数
- auth (AuthSchema): 认证信息模型
返回:
- JSONResponse: 包含查询结果和分页信息的JSON响应
"""
result_dict_list = await GenTableService.get_gen_db_table_list_service(auth=auth, search=search)
result_dict = await PaginationService.paginate(data_list=result_dict_list, page_no=page.page_no, page_size=page.page_size)
log.info('获取数据库表列表成功')
return SuccessResponse(data=result_dict, msg="获取数据库表列表成功")
@GenRouter.post("/import", summary="导入表结构", description="导入表结构")
async def import_gen_table_controller(
table_names: List[str] = Body(..., description="表名列表"),
auth: AuthSchema = Depends(AuthPermission(["module_generator:gencode:import"])),
) -> JSONResponse:
"""
导入表结构
参数:
- table_names (List[str]): 表名列表
- auth (AuthSchema): 认证信息模型
返回:
- JSONResponse: 包含导入结果和导入的表结构列表的JSON响应
"""
add_gen_table_list = await GenTableService.get_gen_db_table_list_by_name_service(auth, table_names)
result = await GenTableService.import_gen_table_service(auth, add_gen_table_list)
log.info('导入表结构成功')
return SuccessResponse(msg="导入表结构成功", data=result)
@GenRouter.get("/detail/{table_id}", summary="获取业务表详细信息", description="获取业务表详细信息")
async def gen_table_detail_controller(
table_id: int = Path(..., description="业务表ID"),
auth: AuthSchema = Depends(AuthPermission(["module_generator:gencode:query"]))
) -> JSONResponse:
"""
获取业务表详细信息
参数:
- table_id (int): 业务表ID
- auth (AuthSchema): 认证信息模型
返回:
- JSONResponse: 包含业务表详细信息的JSON响应
"""
gen_table_detail_result = await GenTableService.get_gen_table_detail_service(auth, table_id)
log.info(f'获取table_id为{table_id}的信息成功')
return SuccessResponse(data=gen_table_detail_result, msg="获取业务表详细信息成功")
@GenRouter.post("/create", summary="创建表结构", description="创建表结构")
async def create_table_controller(
sql: str = Body(..., description="SQL语句,用于创建表结构"),
auth: AuthSchema = Depends(AuthPermission(["module_generator:gencode:create"])),
) -> JSONResponse:
"""
创建表结构
参数:
- sql (str): SQL语句,用于创建表结构
- auth (AuthSchema): 认证信息模型
返回:
- JSONResponse: 包含创建结果的JSON响应
"""
result = await GenTableService.create_table_service(auth, sql)
log.info('创建表结构成功')
return SuccessResponse(msg="创建表结构成功", data=result)
@GenRouter.put("/update/{table_id}", summary="编辑业务表信息", description="编辑业务表信息")
async def update_gen_table_controller(
table_id: int = Path(..., description="业务表ID"),
data: GenTableSchema = Body(..., description="业务表信息"),
auth: AuthSchema = Depends(AuthPermission(["module_generator:gencode:update"])),
) -> JSONResponse:
"""
编辑业务表信息
参数:
- table_id (int): 业务表ID
- data (GenTableSchema): 业务表信息模型
- auth (AuthSchema): 认证信息模型
返回:
- JSONResponse: 包含编辑结果的JSON响应
"""
result_dict = await GenTableService.update_gen_table_service(auth, data, table_id)
log.info('编辑业务表信息成功')
return SuccessResponse(data=result_dict, msg="编辑业务表信息成功")
@GenRouter.delete("/delete", summary="删除业务表信息", description="删除业务表信息")
async def delete_gen_table_controller(
ids: List[int] = Body(..., description="业务表ID列表"),
auth: AuthSchema = Depends(AuthPermission(["module_generator:gencode:delete"]))
) -> JSONResponse:
"""
删除业务表信息
参数:
- ids (List[int]): 业务表ID列表
- auth (AuthSchema): 认证信息模型
返回:
- JSONResponse: 包含删除结果的JSON响应
"""
result = await GenTableService.delete_gen_table_service(auth, ids)
log.info('删除业务表信息成功')
return SuccessResponse(msg="删除业务表信息成功", data=result)
@GenRouter.patch("/batch/output", summary="批量生成代码", description="批量生成代码")
async def batch_gen_code_controller(
table_names: List[str] = Body(..., description="表名列表"),
auth: AuthSchema = Depends(AuthPermission(["module_generator:gencode:patch"]))
) -> StreamResponse:
"""
批量生成代码
参数:
- table_names (List[str]): 表名列表
- auth (AuthSchema): 认证信息模型
返回:
- StreamResponse: 包含批量生成代码的ZIP文件流响应
"""
batch_gen_code_result = await GenTableService.batch_gen_code_service(auth, table_names)
log.info(f'批量生成代码成功,表名列表:{table_names}')
return StreamResponse(
data=bytes2file_response(batch_gen_code_result),
media_type='application/zip',
headers={'Content-Disposition': 'attachment; filename=code.zip'}
)
@GenRouter.post("/output/{table_name}", summary="生成代码到指定路径", description="生成代码到指定路径")
async def gen_code_local_controller(
table_name: str = Path(..., description="表名"),
auth: AuthSchema = Depends(AuthPermission(["module_generator:gencode:code"]))
) -> JSONResponse:
"""
生成代码到指定路径
参数:
- table_name (str): 表名
- auth (AuthSchema): 认证信息模型
返回:
- JSONResponse: 包含生成结果的JSON响应
"""
result = await GenTableService.generate_code_service(auth, table_name)
log.info(f'生成代码,表名:{table_name},到指定路径成功')
return SuccessResponse(msg="生成代码到指定路径成功", data=result)
@GenRouter.get("/preview/{table_id}", summary="预览代码", description="预览代码")
async def preview_code_controller(
table_id: int = Path(..., description="业务表ID"),
auth: AuthSchema = Depends(AuthPermission(["module_generator:gencode:query"]))
) -> JSONResponse:
"""
预览代码
参数:
- table_id (int): 业务表ID
- auth (AuthSchema): 认证信息模型
返回:
- JSONResponse: 包含预览代码的JSON响应
"""
preview_code_result = await GenTableService.preview_code_service(auth, table_id)
log.info(f'预览代码,表id{table_id},成功')
return SuccessResponse(data=preview_code_result, msg="预览代码成功")
@GenRouter.post("/sync_db/{table_name}", summary="同步数据库", description="同步数据库")
async def sync_db_controller(
table_name: str = Path(..., description="表名"),
auth: AuthSchema = Depends(AuthPermission(["module_generator:db:sync"]))
) -> JSONResponse:
"""
同步数据库
参数:
- table_name (str): 表名
- auth (AuthSchema): 认证信息模型
返回:
- JSONResponse: 包含同步数据库结果的JSON响应
"""
result = await GenTableService.sync_db_service(auth, table_name)
log.info(f'同步数据库,表名:{table_name},成功')
return SuccessResponse(msg="同步数据库成功", data=result)
@@ -0,0 +1,448 @@
# -*- coding: utf-8 -*-
from sqlalchemy.engine.reflection import Inspector
from sqlalchemy import Inspector, select, text, inspect
from typing import Sequence
from app.core.logger import log
from app.config.setting import settings
from app.core.base_crud import CRUDBase
from app.api.v1.module_system.auth.schema import AuthSchema
from .model import GenTableModel, GenTableColumnModel
from .schema import (
GenTableSchema,
GenTableColumnSchema,
GenTableColumnOutSchema,
GenDBTableSchema,
GenTableQueryParam
)
class GenTableCRUD(CRUDBase[GenTableModel, GenTableSchema, GenTableSchema]):
"""代码生成业务表模块数据库操作层"""
def __init__(self, auth: AuthSchema) -> None:
"""
初始化CRUD操作层
参数:
- auth (AuthSchema): 认证信息模型
"""
super().__init__(model=GenTableModel, auth=auth)
async def get_gen_table_by_id(self, table_id: int, preload: list | None = None) -> GenTableModel | None:
"""
根据业务表ID获取需要生成的业务表信息。
参数:
- table_id (int): 业务表ID。
- preload (list | None): 预加载关系,未提供时使用模型默认项
返回:
- GenTableModel | None: 业务表信息对象。
"""
return await self.get(id=table_id, preload=preload)
async def get_gen_table_by_name(self, table_name: str, preload: list | None = None) -> GenTableModel | None:
"""
根据业务表名称获取需要生成的业务表信息。
参数:
- table_name (str): 业务表名称。
- preload (list | None): 预加载关系,未提供时使用模型默认项
返回:
- GenTableModel | None: 业务表信息对象。
"""
return await self.get(table_name=table_name, preload=preload)
async def get_gen_table_all(self, preload: list | None = None) -> Sequence[GenTableModel]:
"""
获取所有业务表信息。
参数:
- preload (list | None): 预加载关系,未提供时使用模型默认项
返回:
- Sequence[GenTableModel]: 所有业务表信息列表。
"""
return await self.list(preload=preload)
async def get_gen_table_list(self, search: GenTableQueryParam | None = None, preload: list | None = None) -> Sequence[GenTableModel]:
"""
根据查询参数获取代码生成业务表列表信息。
参数:
- search (GenTableQueryParam | None): 查询参数对象。
- preload (list | None): 预加载关系,未提供时使用模型默认项
返回:
- Sequence[GenTableModel]: 业务表列表信息。
"""
return await self.list(search=search.__dict__, order_by=[{"created_time": "desc"}], preload=preload)
async def add_gen_table(self, add_model: GenTableSchema) -> GenTableModel:
"""
新增业务表信息。
参数:
- add_model (GenTableSchema): 新增业务表信息模型。
返回:
- GenTableModel: 新增的业务表信息对象。
"""
return await self.create(data=add_model)
async def edit_gen_table(self, table_id: int, edit_model: GenTableSchema) -> GenTableModel:
"""
修改业务表信息。
参数:
- table_id (int): 业务表ID。
- edit_model (GenTableSchema): 修改业务表信息模型。
返回:
- GenTableSchema: 修改后的业务表信息模型。
"""
# 排除嵌套对象字段,避免SQLAlchemy尝试直接将字典设置到模型实例上
return await self.update(id=table_id, data=edit_model.model_dump(exclude_unset=True, exclude={"columns"}))
async def delete_gen_table(self, ids: list[int]) -> None:
"""
删除业务表信息。除了系统表。
参数:
- ids (list[int]): 业务表ID列表。
"""
await self.delete(ids=ids)
async def get_db_table_list(self, search: GenTableQueryParam | None = None) -> list[dict]:
"""
根据查询参数获取数据库表列表信息。
参数:
- search (GenTableQueryParam | None): 查询参数对象。
返回:
- list[dict]: 数据库表列表信息(已转为可序列化字典)。
"""
database_name = settings.DATABASE_NAME
database_type = settings.DATABASE_TYPE
from app.core.database import engine
inspector: Inspector = inspect(engine)
table_names = inspector.get_table_names()
dict_data = []
for table_name in table_names:
try:
table_comment = inspector.get_table_comment(table_name)
comment = table_comment.get('text', '') if isinstance(table_comment, dict) else table_comment
table_comment = comment or ""
except Exception as e:
log.warning(f"获取表 {table_name} 的注释失败: {e}")
table_comment = ""
# 统一处理 search 为 None 的情况,避免重复判断
if search:
# 表名过滤:忽略大小写,支持模糊匹配
if search.table_name and search.table_name.lower() not in table_name.lower():
continue
# 表注释过滤:忽略大小写,支持模糊匹配;table_comment 为 None 时视为空字符串
if search.table_comment and search.table_comment not in table_comment:
continue
table_info = {
"database_name": database_name,
"table_name": table_name,
"table_type": database_type,
"table_comment": table_comment
}
dict_data.append(GenDBTableSchema(**table_info).model_dump())
return dict_data
async def get_db_table_list_by_names(self, table_names: list[str]) -> list[GenDBTableSchema]:
"""
根据业务表名称列表获取数据库表信息。
参数:
- table_names (list[str]): 业务表名称列表。
返回:
- list[GenDBTableSchema]: 数据库表信息对象列表。
"""
# 处理空列表情况
if not table_names:
return []
# 调用get_db_table_list获取所有表信息
all_tables = await self.get_db_table_list()
# 过滤出指定名称的表
table_names_set = set(table_names) # 转换为集合以提高查找效率
filtered_tables = [
GenDBTableSchema(**table)
for table in all_tables
if table["table_name"] in table_names_set
]
return filtered_tables
async def check_table_exists(self, table_name: str) -> bool:
"""
检查数据库中是否已存在指定表名的表。
参数:
- table_name (str): 要检查的表名。
返回:
- bool: 如果表存在返回True,否则返回False。
"""
from app.core.database import engine
inspector: Inspector = inspect(engine)
return inspector.has_table(table_name)
async def execute_sql(self, sql: str) -> bool:
"""
执行SQL语句。
参数:
- sql (str): 要执行的SQL语句。
返回:
- bool: 是否执行成功。
"""
try:
# 执行SQL但不手动提交事务,由框架管理事务生命周期
await self.auth.db.execute(text(sql))
return True
except Exception as e:
log.error(f"执行SQL时发生错误: {e}")
return False
class GenTableColumnCRUD(CRUDBase[GenTableColumnModel, GenTableColumnSchema, GenTableColumnSchema]):
"""代码生成业务表字段模块数据库操作层"""
def __init__(self, auth: AuthSchema) -> None:
"""
初始化CRUD操作层
参数:
- auth (AuthSchema): 认证信息模型
"""
super().__init__(model=GenTableColumnModel, auth=auth)
@staticmethod
def _sync_get_table_columns(database_type, table_name):
"""
同步函数:获取数据库表的列信息
参数:
- database_type: 数据库类型
- table_name: 表名
返回:
- list: 列信息列表
"""
# 使用SQLAlchemy Inspector获取表列信息
from app.core.database import engine
inspector: Inspector = inspect(engine)
# 获取列信息
columns = inspector.get_columns(table_name)
# 获取主键信息
try:
pk_constraint = inspector.get_pk_constraint(table_name)
primary_keys = set(pk_constraint.get("constrained_columns", [])) if pk_constraint else set()
except Exception:
primary_keys = set()
# 获取唯一约束信息
unique_columns = set()
try:
unique_constraints = inspector.get_unique_constraints(table_name)
for constraint in unique_constraints:
unique_columns.update(constraint.get("column_names", []))
except Exception:
pass
# 处理列信息
columns_list = []
for idx, column in enumerate(columns):
# 获取列的基本信息
column_name = column['name']
column_type = str(column['type'])
is_nullable = column.get('nullable', True)
column_default = column.get('default', None)
# 获取列注释(如果有的话)
column_comment = column.get('comment', '')
# 判断是否为主键
is_pk = column_name in primary_keys
# 判断是否为唯一约束
is_unique = column_name in unique_columns
# 判断是否为自增列(基于数据库类型和列类型)
is_increment = column.get('autoincrement', False) in (True, 'auto')
# 获取列长度(如果适用)
column_length = None
# 使用getattr安全地获取length属性,避免访问不存在时抛出AttributeError
column_length = getattr(column['type'], 'length', None)
if column_length is not None:
column_length = str(getattr(column['type'], 'length', ''))
# 构造列信息字典
column_info = {
"column_name": column_name,
"column_comment": column_comment or '',
"column_type": column_type,
"column_length": column_length or '',
"column_default": str(column_default) if column_default is not None else '',
"sort": idx + 1, # 序号从1开始
"is_pk": 1 if is_pk else 0,
"is_increment": 1 if is_increment else 0,
"is_nullable": 1 if is_nullable else 0,
"is_unique": 1 if is_unique else 0
}
columns_list.append(column_info)
return columns_list
async def get_gen_table_column_by_id(self, id: int, preload: list | None = None) -> GenTableColumnModel | None:
"""根据业务表字段ID获取业务表字段信息。
参数:
- id (int): 业务表字段ID。
- preload (list | None): 预加载关系,未提供时使用模型默认项
返回:
- GenTableColumnModel | None: 业务表字段信息对象。
"""
return await self.get(id=id, preload=preload)
async def get_gen_table_column_list_by_table_id(self, table_id: int, preload: list | None = None) -> GenTableColumnModel | None:
"""根据业务表ID获取业务表字段列表信息。
参数:
- table_id (int): 业务表ID。
- preload (list | None): 预加载关系,未提供时使用模型默认项
返回:
- GenTableColumnModel | None: 业务表字段列表信息对象。
"""
return await self.get(table_id=table_id, preload=preload)
async def list_gen_table_column_crud_by_table_id(self, table_id: int, order_by: list | None = None, preload: list | None = None) -> Sequence[GenTableColumnModel]:
"""根据业务表ID查询业务表字段列表。
参数:
- table_id (int): 业务表ID。
- order_by (list | None): 排序字段列表,每个元素为{"field": "字段名", "order": "asc" | "desc"}。
- preload (list | None): 预加载关系,未提供时使用模型默认项
返回:
- Sequence[GenTableColumnModel]: 业务表字段列表信息对象序列。
"""
return await self.list(search={"table_id": table_id}, order_by=order_by, preload=preload)
async def get_gen_db_table_columns_by_name(self, table_name: str | None) -> list[GenTableColumnOutSchema]:
"""
根据业务表名称获取业务表字段列表信息。
参数:
- table_name (str | None): 业务表名称。
返回:
- list[GenTableColumnOutSchema]: 业务表字段列表信息对象。
"""
# 检查表名是否为空
if not table_name:
raise ValueError("数据表名称不能为空")
try:
# 直接调用同步方法获取列信息
columns_info = GenTableColumnCRUD._sync_get_table_columns(
settings.DATABASE_TYPE,
table_name
)
# 转换为GenTableColumnOutSchema对象列表
columns_list = []
for column_info in columns_info:
columns_list.append(GenTableColumnOutSchema(**column_info))
return columns_list
except Exception as e:
log.error(f"获取表{table_name}的字段列表时出错: {str(e)}")
# 确保即使出错也返回空列表而不是None
raise
async def list_gen_table_column_crud(self, search: dict | None = None, order_by: list | None = None, preload: list | None = None) -> Sequence[GenTableColumnModel]:
"""根据业务表字段查询业务表字段列表。
参数:
- search (dict | None): 查询参数,例如{"table_id": 1}。
- order_by (list | None): 排序字段列表,每个元素为{"field": "字段名", "order": "asc" | "desc"}。
- preload (list | None): 预加载关系,未提供时使用模型默认项
返回:
- Sequence[GenTableColumnModel]: 业务表字段列表信息对象序列。
"""
return await self.list(search=search, order_by=order_by, preload=preload)
async def create_gen_table_column_crud(self, data: GenTableColumnSchema) -> GenTableColumnModel | None:
"""创建业务表字段。
参数:
- data (GenTableColumnSchema): 业务表字段模型。
返回:
- GenTableColumnModel | None: 业务表字段列表信息对象。
"""
return await self.create(data=data)
async def update_gen_table_column_crud(self, id: int, data: GenTableColumnSchema) -> GenTableColumnModel | None:
"""更新业务表字段。
参数:
- id (int): 业务表字段ID。
- data (GenTableColumnSchema): 业务表字段模型。
返回:
- GenTableColumnModel | None: 业务表字段列表信息对象。
"""
# 将对象转换为字典,避免SQLAlchemy直接操作对象时出现的状态问题
data_dict = data.model_dump(exclude_unset=True)
return await self.update(id=id, data=data_dict)
async def delete_gen_table_column_by_table_id_crud(self, table_ids: list[int]) -> None:
"""根据业务表ID批量删除业务表字段。
参数:
- table_ids (list[int]): 业务表ID列表。
返回:
- None
"""
# 先查询出这些表ID对应的所有字段ID
query = select(GenTableColumnModel.id).where(GenTableColumnModel.table_id.in_(table_ids))
result = await self.auth.db.execute(query)
column_ids = [row[0] for row in result.fetchall()]
# 如果有字段ID,则删除这些字段
if column_ids:
await self.delete(ids=column_ids)
async def delete_gen_table_column_by_column_id_crud(self, column_ids: list[int]) -> None:
"""根据业务表字段ID批量删除业务表字段。
参数:
- column_ids (list[int]): 业务表字段ID列表。
返回:
- None
"""
return await self.delete(ids=column_ids)
@@ -0,0 +1,132 @@
# -*- coding: utf-8 -*-
from sqlalchemy import String, Integer, ForeignKey, Boolean
from sqlalchemy.orm import Mapped, mapped_column, relationship, validates
from sqlalchemy.sql import expression
from app.config.setting import settings
from app.core.base_model import ModelMixin, UserMixin
from app.utils.common_util import SqlalchemyUtil
class GenTableModel(ModelMixin, UserMixin):
"""
代码生成表
"""
__tablename__: str = 'gen_table'
__table_args__: dict[str, str] = ({'comment': '代码生成表'})
__loader_options__: list[str] = ["columns", "created_by", "updated_by"]
table_name: Mapped[str] = mapped_column(String(200), nullable=False, default='', comment='表名称')
table_comment: Mapped[str | None] = mapped_column(String(500), nullable=True, comment='表描述')
class_name: Mapped[str] = mapped_column(String(100), nullable=False, default='', comment='实体类名称')
package_name: Mapped[str | None] = mapped_column(String(100), nullable=True, comment='生成包路径')
module_name: Mapped[str | None] = mapped_column(String(30), nullable=True, comment='生成模块名')
business_name: Mapped[str | None] = mapped_column(String(30), nullable=True, comment='生成业务名')
function_name: Mapped[str | None] = mapped_column(String(100), nullable=True, comment='生成功能名')
sub_table_name: Mapped[str | None] = mapped_column(
String(64),
nullable=True,
server_default=SqlalchemyUtil.get_server_default_null(settings.DATABASE_TYPE),
comment='关联子表的表名'
)
sub_table_fk_name: Mapped[str | None] = mapped_column(
String(64),
nullable=True,
server_default=SqlalchemyUtil.get_server_default_null(settings.DATABASE_TYPE),
comment='子表关联的外键名'
)
parent_menu_id: Mapped[int | None] = mapped_column(Integer, nullable=True, comment='父菜单ID')
# 关联关系
columns: Mapped[list['GenTableColumnModel']] = relationship(
order_by='GenTableColumnModel.sort',
back_populates='table',
cascade='all, delete-orphan'
)
@validates('table_name')
def validate_table_name(self, key: str, table_name: str) -> str:
"""验证表名不为空"""
if not table_name or not table_name.strip():
raise ValueError('表名称不能为空')
return table_name.strip()
@validates('class_name')
def validate_class_name(self, key: str, class_name: str) -> str:
"""验证类名不为空"""
if not class_name or not class_name.strip():
raise ValueError('实体类名称不能为空')
return class_name.strip()
class GenTableColumnModel(ModelMixin, UserMixin):
"""
代码生成表字段
数据隔离策略:
- 继承自GenTableModel的隔离级别
- 不需要customer_id
用于存储代码生成器的字段配置
"""
__tablename__: str = 'gen_table_column'
__table_args__: dict[str, str] = ({'comment': '代码生成表字段'})
__loader_options__: list[str] = ["created_by", "updated_by"]
# 数据库设计表字段
column_name: Mapped[str] = mapped_column(String(200), nullable=False, comment='列名称')
column_comment: Mapped[str | None] = mapped_column(String(500), nullable=True, comment='列描述')
column_type: Mapped[str] = mapped_column(String(100), nullable=False, comment='列类型')
column_length: Mapped[str | None] = mapped_column(String(50), nullable=True, comment='列长度')
column_default: Mapped[str | None] = mapped_column(String(200), nullable=True, comment='列默认值')
is_pk: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False, server_default=expression.false(), comment='是否主键')
is_increment: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False, server_default=expression.false(), comment='是否自增')
is_nullable: Mapped[bool] = mapped_column(Boolean, nullable=False, default=True, server_default=expression.true(), comment='是否允许为空')
is_unique: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False, server_default=expression.false(), comment='是否唯一')
# Python字段映射
python_type: Mapped[str | None] = mapped_column(String(100), nullable=True, comment='Python类型')
python_field: Mapped[str | None] = mapped_column(String(200), nullable=True, comment='Python字段名')
# 序列化配置
is_insert: Mapped[bool] = mapped_column(Boolean, nullable=False, default=True, server_default=expression.true(), comment='是否为新增字段')
is_edit: Mapped[bool] = mapped_column(Boolean, nullable=False, default=True, server_default=expression.true(), comment='是否编辑字段')
is_list: Mapped[bool] = mapped_column(Boolean, nullable=False, default=True, server_default=expression.true(), comment='是否列表字段')
is_query: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False, server_default=expression.false(), comment='是否查询字段')
query_type: Mapped[str | None] = mapped_column(String(50), nullable=True, default=None, comment='查询方式')
# 前端展示配置
html_type: Mapped[str | None] = mapped_column(String(100), nullable=True, default='input', comment='显示类型')
dict_type: Mapped[str | None] = mapped_column(String(200), nullable=True, default='', comment='字典类型')
# 排序和扩展配置
sort: Mapped[int] = mapped_column(Integer, nullable=False, default=0, comment='排序')
# 归属关系
table_id: Mapped[int] = mapped_column(
Integer,
ForeignKey('gen_table.id', ondelete='CASCADE'),
nullable=False,
index=True,
comment='归属表编号'
)
# 关联关系
table: Mapped['GenTableModel'] = relationship(back_populates='columns')
@validates('column_name')
def validate_column_name(self, key: str, column_name: str) -> str:
"""验证列名不为空"""
if not column_name or not column_name.strip():
raise ValueError('列名称不能为空')
return column_name.strip()
@validates('column_type')
def validate_column_type(self, key: str, column_type: str) -> str:
"""验证列类型不为空"""
if not column_type or not column_type.strip():
raise ValueError('列类型不能为空')
return column_type.strip()
@@ -0,0 +1,127 @@
# -*- coding: utf-8 -*-
from pydantic import BaseModel, ConfigDict, Field, field_validator
from fastapi import Query
from app.core.base_schema import BaseSchema
class GenDBTableSchema(BaseModel):
"""数据库中的表信息(跨方言统一结构)。
- 供“导入表结构”与“同步结构”环节使用。
"""
model_config = ConfigDict(from_attributes=True)
database_name: str | None = Field(default=None, description='数据库名称')
table_name: str | None = Field(default=None, description='表名称')
table_type: str | None = Field(default=None, description='表类型')
table_comment: str | None = Field(default=None, description='表描述')
class GenTableColumnSchema(BaseModel):
"""代码生成业务表字段创建模型(原始字段+生成配置)。
- 从根本上解决问题:所有字段都设置了合理的默认值,避免None值问题
"""
model_config = ConfigDict(from_attributes=True)
table_id: int = Field(default=0, description='归属表编号')
column_name: str = Field(default='', description='列名称')
column_comment: str | None = Field(default='', description='列描述')
column_type: str = Field(default='varchar(255)', description='列类型')
column_length: str | None = Field(default='', description='列长度')
column_default: str | None = Field(default='', description='列默认值')
is_pk: bool = Field(default=False, description='是否主键(True是 False否)')
is_increment: bool = Field(default=False, description='是否自增(True是 False否)')
is_nullable: bool = Field(default=True, description='是否允许为空(True是 False否)')
is_unique: bool = Field(default=False, description='是否唯一(True是 False否)')
python_type: str = Field(default='str', description='python类型')
python_field: str = Field(default='', description='python字段名')
is_insert: bool = Field(default=True, description='是否为插入字段(True是 False否)')
is_edit: bool = Field(default=True, description='是否编辑字段(True是 False否)')
is_list: bool = Field(default=True, description='是否列表字段(True是 False否)')
is_query: bool = Field(default=True, description='是否查询字段(True是 False否)')
query_type: str | None = Field(default=None, description='查询方式(等于、不等于、大于、小于、范围)')
html_type: str | None = Field(default='input', description='显示类型(文本框、文本域、下拉框、复选框、单选框、日期控件)')
dict_type: str | None = Field(default='', description='字典类型')
sort: int = Field(default=0, description='排序')
class GenTableColumnOutSchema(GenTableColumnSchema, BaseSchema):
"""
业务表字段输出模型
"""
model_config = ConfigDict(from_attributes=True)
super_column: str | None = Field(default='0', description='是否为基类字段(1是 0否)')
class GenTableSchema(BaseModel):
"""代码生成业务表更新模型(扩展聚合字段)。
- 聚合:`columns`字段包含字段列表;`pk_column`主键字段;子表结构`sub_table`。
"""
"""代码生成业务表基础模型(创建/更新共享字段)。
- 说明:`params`为前端结构体,后端持久化为`options`的JSON。
"""
model_config = ConfigDict(from_attributes=True)
table_name: str= Field(..., description='表名称')
table_comment: str | None = Field(default=None, description='表描述')
class_name: str | None = Field(default=None, description='实体类名称')
package_name: str | None = Field(default=None, description='生成包路径')
module_name: str | None = Field(default=None, description='生成模块名')
business_name: str | None = Field(default=None, description='生成业务名')
function_name: str | None = Field(default=None, description='生成功能名')
sub_table_name: str | None = Field(default=None, description='关联子表的表名')
sub_table_fk_name: str | None = Field(default=None, description='子表关联的外键名')
parent_menu_id: int | None = Field(default=None, description='所属父级分类,生成页面时候生成菜单使用')
description: str | None = Field(default=None, max_length=255, description="描述")
columns: list['GenTableColumnOutSchema'] | None = Field(default=None, description='表列信息')
@field_validator('table_name')
@classmethod
def table_name_update(cls, v: str) -> str:
"""更新表名称"""
if not v:
raise ValueError('表名称不能为空')
return v
class GenTableOutSchema(GenTableSchema, BaseSchema):
"""业务表输出模型(面向控制器/前端)。
"""
model_config = ConfigDict(from_attributes=True)
pk_column: GenTableColumnOutSchema | None = Field(default=None, description='主键信息')
sub_table: GenTableSchema | None = Field(default=None, description='子表信息')
sub: bool | None = Field(default=None, description='是否为子表')
class GenTableQueryParam:
"""代码生成业务表查询参数
- 支持按`table_name`、`table_comment`进行模糊检索(由CRUD层实现like)。
- 空值将被忽略,不参与过滤。
"""
def __init__(
self,
table_name: str | None = Query(None, description="表名称"),
table_comment: str | None = Query(None, description="表注释"),
) -> None:
# 模糊查询字段
self.table_name = table_name
self.table_comment = table_comment
class GenTableColumnQueryParam:
"""代码生成业务表字段查询参数
- `column_name`按like规则模糊查询(透传到CRUD层)
"""
def __init__(
self,
column_name: str | None = Query(None, description="列名称"),
) -> None:
# 模糊查询字段:约定("like", 值)格式,便于CRUD解析
self.column_name = ("like", column_name)
@@ -0,0 +1,719 @@
# -*- coding: utf-8 -*-
import io
import os
import zipfile
from typing import Any
from sqlglot.expressions import Add, Alter, Create, Delete, Drop, Expression, Insert, Table, TruncateTable, Update
import sqlglot
from app.config.path_conf import BASE_DIR
from app.config.setting import settings
from app.core.logger import log
from app.core.exceptions import CustomException
from app.api.v1.module_system.auth.schema import AuthSchema
from .tools.jinja2_template_util import Jinja2TemplateUtil
from .tools.gen_util import GenUtils
from .schema import GenTableSchema, GenTableOutSchema, GenTableColumnSchema, GenTableColumnOutSchema, GenTableQueryParam
from .crud import GenTableColumnCRUD, GenTableCRUD
def handle_service_exception(func):
async def wrapper(*args, **kwargs):
try:
return await func(*args, **kwargs)
except CustomException:
raise
except Exception as e:
raise CustomException(msg=f'{func.__name__}执行失败: {str(e)}')
return wrapper
class GenTableService:
"""代码生成业务表服务层"""
@classmethod
@handle_service_exception
async def get_gen_table_detail_service(cls, auth: AuthSchema, table_id: int) -> dict:
"""获取详细信息。
参数:
- auth (AuthSchema): 认证信息。
- table_id (int): 业务表ID。
返回:
- dict: 包含业务表详细信息的字典。
"""
gen_table = await cls.get_gen_table_by_id_service(auth, table_id)
return GenTableOutSchema.model_validate(gen_table).model_dump()
@classmethod
@handle_service_exception
async def get_gen_table_list_service(cls, auth: AuthSchema, search: GenTableQueryParam) -> list[dict]:
"""
获取代码生成业务表列表信息。
参数:
- auth (AuthSchema): 认证信息。
- search (GenTableQueryParam): 查询参数模型。
返回:
- list[dict]: 包含业务表列表信息的字典列表。
"""
gen_table_list_result = await GenTableCRUD(auth=auth).get_gen_table_list(search)
return [GenTableOutSchema.model_validate(obj).model_dump() for obj in gen_table_list_result]
@classmethod
@handle_service_exception
async def get_gen_db_table_list_service(cls, auth: AuthSchema, search: GenTableQueryParam) -> list[Any]:
"""获取数据库表列表。
参数:
- auth (AuthSchema): 认证信息。
- search (GenTableQueryParam): 查询参数模型。
返回:
- list[Any]: 包含数据库表列表信息的任意类型列表。
"""
gen_db_table_list_result = await GenTableCRUD(auth=auth).get_db_table_list(search)
return gen_db_table_list_result
@classmethod
@handle_service_exception
async def get_gen_db_table_list_by_name_service(cls, auth: AuthSchema, table_names: list[str]) -> list[GenTableOutSchema]:
"""根据表名称组获取数据库表信息。
参数:
- auth (AuthSchema): 认证信息。
- table_names (list[str]): 业务表名称列表。
返回:
- list[GenTableOutSchema]: 包含业务表详细信息的模型列表。
"""
gen_db_table_list_result = await GenTableCRUD(auth).get_db_table_list_by_names(table_names)
# 修复:将GenDBTableSchema对象转换为字典后再传递给GenTableOutSchema
result = []
for gen_table in gen_db_table_list_result:
result.append(GenTableOutSchema(**gen_table.model_dump()))
return result
@classmethod
@handle_service_exception
async def import_gen_table_service(cls, auth: AuthSchema, gen_table_list: list[GenTableOutSchema]) -> bool:
"""导入表结构到生成器。
参数:
- auth (AuthSchema): 认证信息。
- gen_table_list (list[GenTableOutSchema]): 包含业务表详细信息的模型列表。
返回:
- bool: 成功时返回True,失败时抛出异常。
"""
# 检查是否有表需要导入
if not gen_table_list:
raise CustomException(msg="导入的表结构不能为空")
try:
for table in gen_table_list:
table_name = table.table_name
# 检查表是否已存在
existing_table = await GenTableCRUD(auth).get_gen_table_by_name(table_name)
if existing_table:
raise CustomException(msg=f"以下表已存在,不能重复导入: {table_name}")
GenUtils.init_table(table)
if not table.columns:
table.columns = []
add_gen_table = await GenTableCRUD(auth).add_gen_table(GenTableSchema.model_validate(table.model_dump()))
gen_table_columns = await GenTableColumnCRUD(auth).get_gen_db_table_columns_by_name(table_name)
if len(gen_table_columns) > 0:
table.id = add_gen_table.id
for column in gen_table_columns:
column_schema = GenTableColumnSchema(
table_id=table.id,
column_name=column.column_name,
column_comment=column.column_comment,
column_type=column.column_type,
column_length=column.column_length,
column_default=column.column_default,
is_pk=column.is_pk,
is_increment=column.is_increment,
is_nullable=column.is_nullable,
is_unique=column.is_unique,
sort=column.sort,
python_type=column.python_type,
python_field=column.python_field,
)
GenUtils.init_column_field(column_schema, table)
await GenTableColumnCRUD(auth).create_gen_table_column_crud(column_schema)
return True
except Exception as e:
raise CustomException(msg=f'导入失败, {str(e)}')
@classmethod
@handle_service_exception
async def create_table_service(cls, auth: AuthSchema, sql: str) -> bool | None:
"""创建表结构并导入至代码生成模块。
参数:
- auth (AuthSchema): 认证信息。
- sql (str): 包含`CREATE TABLE`语句的SQL字符串。
返回:
- bool | None: 成功时返回True,失败时抛出异常。
"""
# 验证SQL非空
if not sql or not sql.strip():
raise CustomException(msg='SQL语句不能为空')
try:
# 解析SQL语句
sql_statements = sqlglot.parse(sql, dialect=settings.DATABASE_TYPE)
if not sql_statements:
raise CustomException(msg='无法解析SQL语句,请检查SQL语法')
# 校验sql语句是否为合法的建表语句
validate_create = [isinstance(sql_statement, Create) for sql_statement in sql_statements]
validate_forbidden_keywords = [
isinstance(
sql_statement,
(Add, Alter, Delete, Drop, Insert, TruncateTable, Update),
)
for sql_statement in sql_statements
]
if not any(validate_create) or any(validate_forbidden_keywords):
raise CustomException(msg='sql语句不是合法的建表语句')
# 获取要创建的表名
table_names = []
for sql_statement in sql_statements:
if isinstance(sql_statement, Create):
table = sql_statement.find(Table)
if table and table.name:
table_names.append(table.name)
table_names = list(set(table_names))
# 创建CRUD实例
gen_table_crud = GenTableCRUD(auth=auth)
# 检查每个表是否已存在
for table_name in table_names:
# 检查数据库中是否已存在该表
if await gen_table_crud.check_table_exists(table_name):
raise CustomException(msg=f'{table_name} 已存在,请检查并修改表名后重试')
# 检查代码生成模块中是否已导入该表
existing_table = await gen_table_crud.get_gen_table_by_name(table_name)
if existing_table:
raise CustomException(msg=f'{table_name} 已在代码生成模块中存在,请检查并修改表名后重试')
# 表不存在,执行SQL语句创建表
for sql_statement in sql_statements:
if not isinstance(sql_statement, Create):
continue
exc_sql = sql_statement.sql(dialect=settings.DATABASE_TYPE)
log.info(f'执行SQL语句: {exc_sql}')
if not await gen_table_crud.execute_sql(exc_sql):
raise CustomException(msg=f'执行SQL语句 {exc_sql} 失败,请检查数据库')
return True
except Exception as e:
raise CustomException(msg=f'创建表结构失败: {str(e)}')
@classmethod
@handle_service_exception
async def update_gen_table_service(cls, auth: AuthSchema, data: GenTableSchema, table_id: int) -> dict[str, Any]:
"""编辑业务表信息。
参数:
- auth (AuthSchema): 认证信息。
- data (GenTableSchema): 包含业务表详细信息的模型。
- table_id (int): 业务表ID。
返回:
- dict[str, Any]: 更新后的业务表信息。
"""
# 处理params为None的情况
gen_table_info = await cls.get_gen_table_by_id_service(auth, table_id)
if gen_table_info.id:
try:
# 直接调用edit_gen_table方法,它会在内部处理排除嵌套字段的逻辑
result = await GenTableCRUD(auth).edit_gen_table(table_id, data)
# 处理data.columns为None的情况
if data.columns:
for gen_table_column in data.columns:
# 确保column有id字段
if hasattr(gen_table_column, 'id') and gen_table_column.id:
column_schema = GenTableColumnSchema(**gen_table_column.model_dump())
await GenTableColumnCRUD(auth).update_gen_table_column_crud(gen_table_column.id, column_schema)
return GenTableOutSchema.model_validate(result).model_dump()
except Exception as e:
raise CustomException(msg=str(e))
else:
raise CustomException(msg='业务表不存在')
@classmethod
@handle_service_exception
async def delete_gen_table_service(cls, auth: AuthSchema, ids: list[int]) -> None:
"""删除业务表信息(先删字段,再删表)。
参数:
- auth (AuthSchema): 认证信息。
- ids (list[int]): 业务表ID列表。
返回:
- None
"""
# 验证ID列表非空
if not ids:
raise CustomException(msg="ID列表不能为空")
try:
# 先删除相关的字段信息
await GenTableColumnCRUD(auth=auth).delete_gen_table_column_by_table_id_crud(ids)
# 再删除表信息
await GenTableCRUD(auth=auth).delete_gen_table(ids)
except Exception as e:
raise CustomException(msg=str(e))
@classmethod
@handle_service_exception
async def get_gen_table_by_id_service(cls, auth: AuthSchema, table_id: int) -> GenTableOutSchema:
"""获取需要生成代码的业务表详细信息。
参数:
- auth (AuthSchema): 认证信息。
- table_id (int): 业务表ID。
返回:
- GenTableOutSchema: 业务表详细信息模型。
"""
gen_table = await GenTableCRUD(auth=auth).get_gen_table_by_id(table_id)
if not gen_table:
raise CustomException(msg='业务表不存在')
result = GenTableOutSchema.model_validate(gen_table)
return result
@classmethod
@handle_service_exception
async def get_gen_table_all_service(cls, auth: AuthSchema) -> list[GenTableOutSchema]:
"""获取所有业务表信息(列表)。
参数:
- auth (AuthSchema): 认证信息。
返回:
- list[GenTableOutSchema]: 业务表详细信息模型列表。
"""
gen_table_all = await GenTableCRUD(auth=auth).get_gen_table_all() or []
result = []
for gen_table in gen_table_all:
try:
table_out = GenTableOutSchema.model_validate(gen_table)
result.append(table_out)
except Exception as e:
log.error(f"转换业务表时出错: {str(e)}")
continue
return result
@classmethod
@handle_service_exception
async def preview_code_service(cls, auth: AuthSchema, table_id: int) -> dict[str, Any]:
"""
预览代码(根据模板渲染内存结果)。
参数:
- auth (AuthSchema): 认证信息。
- table_id (int): 业务表ID。
返回:
- dict[str, Any]: 文件名到渲染内容的映射。
"""
gen_table = GenTableOutSchema.model_validate(
await GenTableCRUD(auth).get_gen_table_by_id(table_id)
)
await cls.set_pk_column(gen_table)
env = Jinja2TemplateUtil.get_env()
context = Jinja2TemplateUtil.prepare_context(gen_table)
template_list = Jinja2TemplateUtil.get_template_list()
preview_code_result = {}
for template in template_list:
try:
render_content = await env.get_template(template).render_async(**context)
preview_code_result[template] = render_content
except Exception as e:
log.error(f"渲染模板 {template} 时出错: {str(e)}")
# 即使某个模板渲染失败,也继续处理其他模板
preview_code_result[template] = f"渲染错误: {str(e)}"
return preview_code_result
@classmethod
@handle_service_exception
async def generate_code_service(cls, auth: AuthSchema, table_name: str) -> bool:
"""生成代码至指定路径(安全写入+可跳过覆盖)。
- 安全:限制写入在项目根目录内;越界路径自动回退到项目根目录。
参数:
- auth (AuthSchema): 认证信息。
- table_name (str): 业务表名。
返回:
- bool: 生成是否成功。
"""
# 验证表名非空
if not table_name or not table_name.strip():
raise CustomException(msg='表名不能为空')
env = Jinja2TemplateUtil.get_env()
render_info = await cls.__get_gen_render_info(auth, table_name)
gen_table_schema: GenTableOutSchema = render_info[3]
from app.api.v1.module_system.menu.crud import MenuCRUD
from app.api.v1.module_system.menu.schema import MenuCreateSchema
from app.utils.common_util import CamelCaseUtil
# 构建权限前缀
permission_prefix = f"{gen_table_schema.module_name}:{gen_table_schema.business_name}"
# 创建菜单 CRUD 实例
menu_crud = MenuCRUD(auth)
if not gen_table_schema.business_name:
raise CustomException(msg='业务名称不能为空')
if not gen_table_schema.function_name:
raise CustomException(msg='功能名称不能为空')
if not gen_table_schema.package_name:
raise CustomException(msg='包名不能为空')
# 1. 先检查并创建菜单(目录菜单、功能菜单、按钮权限)
# 检查是否需要创建目录菜单
if gen_table_schema.parent_menu_id:
# 如果传了上级菜单ID(菜单类型=1),则不创建目录菜单,直接使用该ID作为功能菜单的父ID
dir_menu_id = gen_table_schema.parent_menu_id
else:
# 如果没传上级菜单ID,则需要创建新的模块目录菜单(类型=1:目录)
existing_dir_menu = await menu_crud.get(name=gen_table_schema.business_name)
if existing_dir_menu:
dir_menu_id = existing_dir_menu.id
else:
dir_parent_menu = await menu_crud.create(
MenuCreateSchema(
name=gen_table_schema.package_name,
type=1,
order=9999,
permission=None,
icon="menu",
route_name=CamelCaseUtil.snake_to_camel(gen_table_schema.package_name),
route_path=f"/{gen_table_schema.package_name}",
component_path=None,
redirect=f"/{gen_table_schema.package_name}/{gen_table_schema.business_name}",
hidden=False,
keep_alive=True,
always_show=False,
title=gen_table_schema.package_name,
params=None,
affix=False,
parent_id=gen_table_schema.parent_menu_id, # 这里应该是None,因为上面已经判断过了
status="0",
description=f"{gen_table_schema.business_name}目录"
)
)
dir_menu_id = dir_parent_menu.id
# 检查功能菜单是否已存在,如果存在则抛出错误
existing_func_menu = await menu_crud.get(name=gen_table_schema.function_name, type=2)
if existing_func_menu:
raise CustomException(msg=f"功能菜单名称 '{gen_table_schema.function_name}' 已存在,不能重复创建")
else:
# 创建功能菜单(类型=2:菜单)
parent_menu = await menu_crud.create(
MenuCreateSchema(
name=gen_table_schema.function_name,
type=2,
order=9999,
permission=f"{permission_prefix}:query",
icon="menu",
route_name=CamelCaseUtil.snake_to_camel(gen_table_schema.business_name),
route_path=f"/{gen_table_schema.package_name}/{gen_table_schema.business_name}",
component_path=f"{gen_table_schema.module_name}/{gen_table_schema.business_name}/index",
redirect=None,
hidden=False,
keep_alive=True,
always_show=False,
title=gen_table_schema.function_name,
params=None,
affix=False,
parent_id=dir_menu_id, # 使用目录菜单ID或用户指定的parent_menu_id作为父ID
status="0",
description=f"{gen_table_schema.function_name}功能菜单"
)
)
# 创建按钮权限(类型=3:按钮/权限)
buttons = [
{
"name": f"{gen_table_schema.function_name}查询",
"permission": f"{permission_prefix}:query",
"order": 1
},
{
"name": f"{gen_table_schema.function_name}详情",
"permission": f"{permission_prefix}:detail",
"order": 2
},
{
"name": f"{gen_table_schema.function_name}新增",
"permission": f"{permission_prefix}:create",
"order": 3
},
{
"name": f"{gen_table_schema.function_name}修改",
"permission": f"{permission_prefix}:update",
"order": 4
},
{
"name": f"{gen_table_schema.function_name}删除",
"permission": f"{permission_prefix}:delete",
"order": 5
},
{
"name": f"{gen_table_schema.function_name}批量状态修改",
"permission": f"{permission_prefix}:patch",
"order": 6
},
{
"name": f"{gen_table_schema.function_name}导出",
"permission": f"{permission_prefix}:export",
"order": 7
},
{
"name": f"{gen_table_schema.function_name}导入",
"permission": f"{permission_prefix}:import",
"order": 8
},
{
"name": f"{gen_table_schema.function_name}下载导入模板",
"permission": f"{permission_prefix}:download",
"order": 9
}
]
for button in buttons:
# 检查按钮权限是否已存在
await menu_crud.create(
MenuCreateSchema(
name=button["name"],
type=3,
order=button["order"],
permission=button["permission"],
icon=None,
route_name=None,
route_path=None,
component_path=None,
redirect=None,
hidden=False,
keep_alive=True,
always_show=False,
title=button["name"],
params=None,
affix=False,
parent_id=parent_menu.id,
status="0",
description=f"{gen_table_schema.function_name}功能按钮"
)
)
log.info(f"成功创建按钮权限: {button['name']}")
log.info(f"成功创建{gen_table_schema.function_name}菜单及按钮权限")
# 2. 菜单创建成功后,再生成页面代码
for template in render_info[0]:
try:
render_content = await env.get_template(template).render_async(**render_info[2])
file_name = Jinja2TemplateUtil.get_file_name(template, gen_table_schema)
full_path = BASE_DIR.parent.joinpath(file_name)
gen_path = str(full_path)
if not gen_path:
raise CustomException(msg='【代码生成】生成路径为空')
# 确保目录存在
os.makedirs(os.path.dirname(gen_path), exist_ok=True)
with open(gen_path, 'w', encoding='utf-8') as f:
f.write(render_content)
module_init_path = BASE_DIR.parent.joinpath(f'backend/app/api/v1/{gen_table_schema.module_name}/__init__.py')
if not module_init_path.exists():
# 创建module_name目录的__init__.py文件
os.makedirs(os.path.dirname(module_init_path), exist_ok=True)
with open(module_init_path, 'w', encoding='utf-8') as f:
f.write('# -*- coding: utf-8 -*-')
except Exception as e:
raise CustomException(msg=f'渲染模板失败,表名:{gen_table_schema.table_name},详细错误信息:{str(e)}')
return True
@classmethod
@handle_service_exception
async def batch_gen_code_service(cls, auth: AuthSchema, table_names: list[str]) -> bytes:
"""
批量生成代码并打包为ZIP。
- 备注:内存生成并压缩,兼容多模板类型;供下载使用。
参数:
- auth (AuthSchema): 认证信息。
- table_names (list[str]): 业务表名列表。
返回:
- bytes: 包含所有生成代码的ZIP文件内容。
"""
# 验证表名列表非空
if not table_names:
raise CustomException(msg="表名列表不能为空")
zip_buffer = io.BytesIO()
with zipfile.ZipFile(zip_buffer, 'w', zipfile.ZIP_DEFLATED) as zip_file:
for table_name in table_names:
if not table_name.strip():
continue
try:
env = Jinja2TemplateUtil.get_env()
render_info = await cls.__get_gen_render_info(auth, table_name)
for template_file, output_file in zip(render_info[0], render_info[1]):
render_content = await env.get_template(template_file).render_async(**render_info[2])
zip_file.writestr(output_file, render_content)
except Exception as e:
log.error(f"批量生成代码时处理表 {table_name} 出错: {str(e)}")
# 继续处理其他表,不中断整个过程
continue
zip_data = zip_buffer.getvalue()
zip_buffer.close()
return zip_data
@classmethod
@handle_service_exception
async def sync_db_service(cls, auth: AuthSchema, table_name: str) -> None:
"""
同步数据库表结构到业务表。
参数:
- auth (AuthSchema): 认证信息。
- table_name (str): 业务表名。
返回:
- None
"""
# 验证表名非空
if not table_name or not table_name.strip():
raise CustomException(msg='表名不能为空')
gen_table = await GenTableCRUD(auth).get_gen_table_by_name(table_name)
if not gen_table:
raise CustomException(msg='业务表不存在')
table = GenTableOutSchema.model_validate(gen_table)
if not table.id:
raise CustomException(msg='业务表ID不能为空')
table_columns = table.columns or []
table_column_map = {column.column_name: column for column in table_columns}
# 确保db_table_columns始终是列表类型,避免None值
db_table_columns = await GenTableColumnCRUD(auth).get_gen_db_table_columns_by_name(table_name) or []
db_table_columns = [col for col in db_table_columns if col is not None]
db_table_column_names = [column.column_name for column in db_table_columns]
try:
for column in db_table_columns:
GenUtils.init_column_field(column, table)
if column.column_name in table_column_map:
prev_column = table_column_map[column.column_name]
if hasattr(prev_column, 'id') and prev_column.id:
column.id = prev_column.id
if hasattr(prev_column, 'dict_type') and prev_column.dict_type:
column.dict_type = prev_column.dict_type
if hasattr(prev_column, 'query_type') and prev_column.query_type:
column.query_type = prev_column.query_type
if hasattr(prev_column, 'html_type') and prev_column.html_type:
column.html_type = prev_column.html_type
is_pk_bool = False
if hasattr(prev_column, 'is_pk'):
if isinstance(prev_column.is_pk, bool):
is_pk_bool = prev_column.is_pk
else:
is_pk_bool = str(prev_column.is_pk) == '1'
if hasattr(prev_column, 'is_nullable') and not is_pk_bool:
column.is_nullable = prev_column.is_nullable
if hasattr(prev_column, 'python_field'):
column.python_field = prev_column.python_field or column.python_field
if hasattr(column, 'id') and column.id:
await GenTableColumnCRUD(auth).update_gen_table_column_crud(column.id, column)
else:
await GenTableColumnCRUD(auth).create_gen_table_column_crud(column)
else:
# 设置table_id以确保新字段能正确关联到表
column.table_id = table.id
await GenTableColumnCRUD(auth).create_gen_table_column_crud(column)
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 hasattr(column, 'id') and column.id:
await GenTableColumnCRUD(auth).delete_gen_table_column_by_column_id_crud([column.id])
except Exception as e:
raise CustomException(msg=f'同步失败: {str(e)}')
@classmethod
async def set_pk_column(cls, gen_table: GenTableOutSchema) -> None:
"""设置主键列信息(主表/子表)。
- 备注:同时兼容`pk`布尔与`is_pk == '1'`字符串两种标识。
参数:
- gen_table (GenTableOutSchema): 业务表详细信息模型。
返回:
- None
"""
if gen_table.columns:
for column in gen_table.columns:
# 修复:确保正确检查主键标识
if getattr(column, 'pk', False) or getattr(column, 'is_pk', '') == '1':
gen_table.pk_column = column
break
# 如果没有找到主键列且有列存在,使用第一个列作为主键
if gen_table.pk_column is None and gen_table.columns:
gen_table.pk_column = gen_table.columns[0]
@classmethod
async def __get_gen_render_info(cls, auth: AuthSchema, table_name: str) -> list[Any]:
"""
获取生成代码渲染模板相关信息。
参数:
- auth (AuthSchema): 认证对象。
- table_name (str): 业务表名称。
返回:
- list[Any]: [模板列表, 输出文件名列表, 渲染上下文, 业务表对象]。
异常:
- CustomException: 当业务表不存在或数据转换失败时抛出。
"""
gen_table_model = await GenTableCRUD(auth=auth).get_gen_table_by_name(table_name)
# 检查表是否存在
if gen_table_model is None:
raise CustomException(msg=f"业务表 {table_name} 不存在")
gen_table = GenTableOutSchema.model_validate(gen_table_model)
await cls.set_pk_column(gen_table)
context = Jinja2TemplateUtil.prepare_context(gen_table)
template_list = Jinja2TemplateUtil.get_template_list()
output_files = [Jinja2TemplateUtil.get_file_name(template, gen_table) for template in template_list]
return [template_list, output_files, context, gen_table]
class GenTableColumnService:
"""代码生成业务表字段服务层"""
@classmethod
@handle_service_exception
async def get_gen_table_column_list_by_table_id_service(cls, auth: AuthSchema, table_id: int) -> list[dict[str, Any]]:
"""获取业务表字段列表信息(输出模型)。
参数:
- auth (AuthSchema): 认证信息。
- table_id (int): 业务表ID。
返回:
- list[dict[str, Any]]: 业务表字段列表,每个元素为字段详细信息字典。
"""
gen_table_column_list_result = await GenTableColumnCRUD(auth).list_gen_table_column_crud({"table_id": table_id})
result = [GenTableColumnOutSchema.model_validate(gen_table_column).model_dump() for gen_table_column in gen_table_column_list_result]
return result
@@ -0,0 +1 @@
# -*- coding: utf-8 -*-
@@ -0,0 +1,123 @@
# -*- coding: utf-8 -*-
from fastapi import APIRouter, Depends, UploadFile, Body, Path, Query
from fastapi.responses import StreamingResponse, JSONResponse
from app.common.response import SuccessResponse, StreamResponse
from app.core.dependencies import AuthPermission
from app.api.v1.module_system.auth.schema import AuthSchema
from app.core.base_params import PaginationQueryParam
from app.utils.common_util import bytes2file_response
from app.core.logger import log
from app.core.base_schema import BatchSetAvailable
from .service import {{ class_name }}Service
from .schema import {{ class_name }}CreateSchema, {{ class_name }}UpdateSchema, {{ class_name }}QueryParam
{{ class_name }}Router = APIRouter(prefix='/{{ business_name }}', tags=["{{ function_name }}模块"])
@{{ class_name }}Router.get("/detail/{id}", summary="获取{{ function_name }}详情", description="获取{{ function_name }}详情")
async def get_{{ business_name }}_detail_controller(
id: int = Path(..., description="ID"),
auth: AuthSchema = Depends(AuthPermission(["{{ permission_prefix }}:query"]))
) -> JSONResponse:
"""获取{{ function_name }}详情接口"""
result_dict = await {{ class_name }}Service.detail_{{ business_name }}_service(auth=auth, id=id)
log.info(f"获取{{ function_name }}详情成功 {id}")
return SuccessResponse(data=result_dict, msg="获取{{ function_name }}详情成功")
@{{ class_name }}Router.get("/list", summary="查询{{ function_name }}列表", description="查询{{ function_name }}列表")
async def get_{{ business_name }}_list_controller(
page: PaginationQueryParam = Depends(),
search: {{ class_name }}QueryParam = Depends(),
auth: AuthSchema = Depends(AuthPermission(["{{ permission_prefix }}:query"]))
) -> JSONResponse:
"""查询{{ function_name }}列表接口(数据库分页)"""
result_dict = await {{ class_name }}Service.page_{{ business_name }}_service(
auth=auth,
page_no=page.page_no if page.page_no is not None else 1,
page_size=page.page_size if page.page_size is not None else 10,
search=search,
order_by=page.order_by
)
log.info("查询{{ function_name }}列表成功")
return SuccessResponse(data=result_dict, msg="查询{{ function_name }}列表成功")
@{{ class_name }}Router.post("/create", summary="创建{{ function_name }}", description="创建{{ function_name }}")
async def create_{{ business_name }}_controller(
data: {{ class_name }}CreateSchema,
auth: AuthSchema = Depends(AuthPermission(["{{ permission_prefix }}:create"]))
) -> JSONResponse:
"""创建{{ function_name }}接口"""
result_dict = await {{ class_name }}Service.create_{{ business_name }}_service(auth=auth, data=data)
log.info("创建{{ function_name }}成功")
return SuccessResponse(data=result_dict, msg="创建{{ function_name }}成功")
@{{ class_name }}Router.put("/update/{id}", summary="修改{{ function_name }}", description="修改{{ function_name }}")
async def update_{{ business_name }}_controller(
data: {{ class_name }}UpdateSchema,
id: int = Path(..., description="ID"),
auth: AuthSchema = Depends(AuthPermission(["{{ permission_prefix }}:update"]))
) -> JSONResponse:
"""修改{{ function_name }}接口"""
result_dict = await {{ class_name }}Service.update_{{ business_name }}_service(auth=auth, id=id, data=data)
log.info("修改{{ function_name }}成功")
return SuccessResponse(data=result_dict, msg="修改{{ function_name }}成功")
@{{ class_name }}Router.delete("/delete", summary="删除{{ function_name }}", description="删除{{ function_name }}")
async def delete_{{ business_name }}_controller(
ids: list[int] = Body(..., description="ID列表"),
auth: AuthSchema = Depends(AuthPermission(["{{ permission_prefix }}:delete"]))
) -> JSONResponse:
"""删除{{ function_name }}接口"""
await {{ class_name }}Service.delete_{{ business_name }}_service(auth=auth, ids=ids)
log.info(f"删除{{ function_name }}成功: {ids}")
return SuccessResponse(msg="删除{{ function_name }}成功")
@{{ class_name }}Router.patch("/available/setting", summary="批量修改{{ function_name }}状态", description="批量修改{{ function_name }}状态")
async def batch_set_available_{{ business_name }}_controller(
data: BatchSetAvailable,
auth: AuthSchema = Depends(AuthPermission(["{{ permission_prefix }}:patch"]))
) -> JSONResponse:
"""批量修改{{ function_name }}状态接口"""
await {{ class_name }}Service.set_available_{{ business_name }}_service(auth=auth, data=data)
log.info(f"批量修改{{ function_name }}状态成功: {data.ids}")
return SuccessResponse(msg="批量修改{{ function_name }}状态成功")
@{{ class_name }}Router.post('/export', summary="导出{{ function_name }}", description="导出{{ function_name }}")
async def export_{{ business_name }}_list_controller(
search: {{ class_name }}QueryParam = Depends(),
auth: AuthSchema = Depends(AuthPermission(["{{ permission_prefix }}:export"]))
) -> StreamingResponse:
"""导出{{ function_name }}接口"""
result_dict_list = await {{ class_name }}Service.list_{{ business_name }}_service(search=search, auth=auth)
export_result = await {{ class_name }}Service.batch_export_{{ business_name }}_service(obj_list=result_dict_list)
log.info('导出{{ function_name }}成功')
return StreamResponse(
data=bytes2file_response(export_result),
media_type='application/vnd.openxmlformats-officedocument.spreadsheetml.sheet',
headers={
'Content-Disposition': 'attachment; filename={{ table_name }}.xlsx'
}
)
@{{ class_name }}Router.post('/import', summary="导入{{ function_name }}", description="导入{{ function_name }}")
async def import_{{ business_name }}_list_controller(
file: UploadFile,
auth: AuthSchema = Depends(AuthPermission(["{{ permission_prefix }}:import"]))
) -> JSONResponse:
"""导入{{ function_name }}接口"""
batch_import_result = await {{ class_name }}Service.batch_import_{{ business_name }}_service(file=file, auth=auth, update_support=True)
log.info("导入{{ function_name }}成功")
return SuccessResponse(data=batch_import_result, msg="导入{{ function_name }}成功")
@{{ class_name }}Router.post('/download/template', summary="获取{{ function_name }}导入模板", description="获取{{ function_name }}导入模板", dependencies=[Depends(AuthPermission(["{{ permission_prefix }}:download"]))])
async def export_{{ business_name }}_template_controller() -> StreamingResponse:
"""获取{{ function_name }}导入模板接口"""
import_template_result = await {{ class_name }}Service.import_template_download_{{ business_name }}_service()
log.info('获取{{ function_name }}导入模板成功')
return StreamResponse(
data=bytes2file_response(import_template_result),
media_type='application/vnd.openxmlformats-officedocument.spreadsheetml.sheet',
headers={'Content-Disposition': 'attachment; filename={{ table_name }}_template.xlsx'}
)
@@ -0,0 +1,123 @@
# -*- coding: utf-8 -*-
from typing import Sequence
from app.core.base_crud import CRUDBase
from app.api.v1.module_system.auth.schema import AuthSchema
from .model import {{ class_name }}Model
from .schema import {{ class_name }}CreateSchema, {{ class_name }}UpdateSchema, {{ class_name }}OutSchema
class {{ class_name }}CRUD(CRUDBase[{{ class_name }}Model, {{ class_name }}CreateSchema, {{ class_name }}UpdateSchema]):
"""{{ function_name }}数据层"""
def __init__(self, auth: AuthSchema) -> None:
"""
初始化CRUD数据层
参数:
- auth (AuthSchema): 认证信息模型
"""
super().__init__(model={{ class_name }}Model, auth=auth)
async def get_by_id_{{ business_name }}_crud(self, id: int, preload: list | None = None) -> {{ class_name }}Model | None:
"""
详情
参数:
- id (int): 对象ID
- preload (list | None): 预加载关系,未提供时使用模型默认项
返回:
- {{ class_name }}Model | None: 模型实例或None
"""
return await self.get(id=id, preload=preload)
async def list_{{ business_name }}_crud(self, search: dict | None = None, order_by: list[dict] | None = None, preload: list | None = None) -> Sequence[{{ class_name }}Model]:
"""
列表查询
参数:
- search (dict | None): 查询参数
- order_by (list[dict] | None): 排序参数,未提供时使用模型默认项
- preload (list | None): 预加载关系,未提供时使用模型默认项
返回:
- Sequence[{{ class_name }}Model]: 模型实例序列
"""
return await self.list(search=search, order_by=order_by, preload=preload)
async def create_{{ business_name }}_crud(self, data: {{ class_name }}CreateSchema) -> {{ class_name }}Model | None:
"""
创建
参数:
- data ({{ class_name }}CreateSchema): 创建模型
返回:
- {{ class_name }}Model | None: 模型实例或None
"""
return await self.create(data=data)
async def update_{{ business_name }}_crud(self, id: int, data: {{ class_name }}UpdateSchema) -> {{ class_name }}Model | None:
"""
更新
参数:
- id (int): 对象ID
- data ({{ class_name }}UpdateSchema): 更新模型
返回:
- {{ class_name }}Model | None: 模型实例或None
"""
return await self.update(id=id, data=data)
async def delete_{{ business_name }}_crud(self, ids: list[int]) -> None:
"""
批量删除
参数:
- ids (list[int]): 对象ID列表
返回:
- None
"""
return await self.delete(ids=ids)
async def set_available_{{ business_name }}_crud(self, ids: list[int], status: str) -> None:
"""
批量设置可用状态
参数:
- ids (list[int]): 对象ID列表
- status (str): 可用状态
返回:
- None
"""
return await self.set(ids=ids, status=status)
async def page_{{ business_name }}_crud(self, offset: int, limit: int, order_by: list[dict] | None = None, search: dict | None = None, preload: list | None = None) -> dict:
"""
分页查询
参数:
- offset (int): 偏移量
- limit (int): 每页数量
- order_by (list[dict] | None): 排序参数,未提供时使用模型默认项
- search (dict | None): 查询参数,未提供时查询所有
- preload (list | None): 预加载关系,未提供时使用模型默认项
返回:
- Dict: 分页数据
"""
order_by_list = order_by or [{'id': 'asc'}]
search_dict = search or {}
return await self.page(
offset=offset,
limit=limit,
order_by=order_by_list,
search=search_dict,
out_schema={{ class_name }}OutSchema,
preload=preload
)
@@ -0,0 +1,31 @@
# -*- coding: utf-8 -*-
{% for model_import in model_import_list %}
{{ model_import }}
{% endfor %}
{% if table.sub %}
from sqlalchemy.orm import relationship
{% endif %}
from sqlalchemy.orm import Mapped, mapped_column
from app.core.base_model import ModelMixin, UserMixin
class {{ class_name }}Model(ModelMixin, UserMixin):
"""
{{ function_name }}表
"""
__tablename__: str = '{{ table_name }}'
__table_args__: dict[str, str] = {'comment': '{{ function_name }}'}
__loader_options__: list[str] = ["created_by", "updated_by"]
{% for column in columns %}
{% if column.column_name not in ['id', 'uuid', 'status', 'description', 'created_time', 'updated_time', 'created_id', 'updated_id'] %}
{% set sqlalchemy_type = column|get_sqlalchemy_type %}
{{ column.column_name }}: Mapped[{{ column.python_type }} | None] = mapped_column({{ sqlalchemy_type }}, {% if column.pk %}primary_key=True, {% endif %}{% if column.increment %}autoincrement=True, {% endif %}{% if column.required or column.pk %}nullable=False{% else %}nullable=True{% endif %}, comment='{{ column.column_comment }}')
{% endif %}
{% endfor %}
{% if table.sub %}
{{ sub_class_name }}_list = relationship('{{ sub_class_name }}', back_populates='{{ business_name }}')
{% endif %}
@@ -0,0 +1,86 @@
# -*- coding: utf-8 -*-
{% if table.sub %}
from typing import List
{% endif %}
from pydantic import BaseModel, ConfigDict, Field
from fastapi import Query
{% for import_stmt in schema_import_list %}
{{ import_stmt }}
{% endfor %}
{% if table.created_time %}
from app.core.validator import DateTimeStr
{% endif %}
from app.core.base_schema import BaseSchema, UserBySchema
class {{ class_name }}CreateSchema(BaseModel):
"""
{{ function_name }}新增模型
"""
{% for column in columns %}
{% if column.column_name not in ['id', 'uuid', 'created_time', 'updated_time', 'created_id', 'updated_id'] %}
{% if column.column_name == 'status' %}
{{ column.column_name }}: {{ column.python_type }} = Field(default="0", description='{{ column.column_comment }}')
{% elif column.column_name == 'description' %}
{{ column.column_name }}: str | None = Field(default=None, max_length=255, description='{{ column.column_comment }}')
{% else %}
{{ column.column_name }}: {{ column.python_type }} = Field(default=..., description='{{ column.column_comment }}')
{% endif %}
{% endif %}
{% endfor %}
class {{ class_name }}UpdateSchema({{ class_name }}CreateSchema):
"""
{{ function_name }}更新模型
"""
...
class {{ class_name }}OutSchema({{ class_name }}CreateSchema, BaseSchema, UserBySchema):
"""
{{ function_name }}响应模型
"""
model_config = ConfigDict(from_attributes=True)
class {{ class_name }}QueryParam:
"""{{ function_name }}查询参数"""
def __init__(
self,
{% for column in columns %}
{% if column.query_type == 'LIKE' %}
{{ column.column_name }}: str | None = Query(None, description="{{ column.column_comment }}"),
{% endif %}
{% endfor %}
{% for column in columns %}
{% if column.query_type == 'EQ' and column.column_name not in ['created_time', 'updated_time'] %}
{{ column.column_name }}: {{ column.python_type }} | None = Query(None, description="{{ column.column_comment }}"),
{% endif %}
{% endfor %}
{% if table.created_time %}
created_time: list[DateTimeStr] | None = Query(None, description="创建时间范围", examples=["2025-01-01 00:00:00", "2025-12-31 23:59:59"]),
{% endif %}
{% if table.updated_time %}
updated_time: list[DateTimeStr] | None = Query(None, description="更新时间范围", examples=["2025-01-01 00:00:00", "2025-12-31 23:59:59"]),
{% endif %}
) -> None:
{% for column in columns %}
{% if column.query_type == 'LIKE' %}
# 模糊查询字段
self.{{ column.column_name }} = ("like", {{ column.column_name }})
{% elif column.query_type == 'EQ' and column.column_name not in ['created_time', 'updated_time'] %}
# 精确查询字段
self.{{ column.column_name }} = {{ column.column_name }}
{% endif %}
{% endfor %}
{% if table.created_time %}
# 时间范围查询
if created_time and len(created_time) == 2:
self.created_time = ("between", (created_time[0], created_time[1]))
{% endif %}
{% if table.updated_time %}
if updated_time and len(updated_time) == 2:
self.updated_time = ("between", (updated_time[0], updated_time[1]))
{% endif %}
@@ -0,0 +1,228 @@
# -*- coding: utf-8 -*-
import io
from fastapi import UploadFile
import pandas as pd
from app.core.base_schema import BatchSetAvailable
from app.core.exceptions import CustomException
from app.utils.excel_util import ExcelUtil
from app.core.logger import log
from app.api.v1.module_system.auth.schema import AuthSchema
from .schema import {{ class_name }}CreateSchema, {{ class_name }}UpdateSchema, {{ class_name }}OutSchema, {{ class_name }}QueryParam
from .crud import {{ class_name }}CRUD
class {{ class_name }}Service:
"""
{{ function_name }}服务层
"""
@classmethod
async def detail_{{ business_name }}_service(cls, auth: AuthSchema, id: int) -> dict:
"""详情"""
obj = await {{ class_name }}CRUD(auth).get_by_id_{{ business_name }}_crud(id=id)
if not obj:
raise CustomException(msg="该数据不存在")
return {{ class_name }}OutSchema.model_validate(obj).model_dump()
@classmethod
async def list_{{ business_name }}_service(cls, auth: AuthSchema, search: {{ class_name }}QueryParam | None = None, order_by: list[dict] | None = None) -> list[dict]:
"""列表查询"""
search_dict = search.__dict__ if search else None
obj_list = await {{ class_name }}CRUD(auth).list_{{ business_name }}_crud(search=search_dict, order_by=order_by)
return [{{ class_name }}OutSchema.model_validate(obj).model_dump() for obj in obj_list]
@classmethod
async def page_{{ business_name }}_service(cls, auth: AuthSchema, page_no: int, page_size: int, search: {{ class_name }}QueryParam | None = None, order_by: list[dict] | None = None) -> dict:
"""分页查询(数据库分页)"""
search_dict = search.__dict__ if search else {}
order_by_list = order_by or [{'id': 'asc'}]
offset = (page_no - 1) * page_size
result = await {{ class_name }}CRUD(auth).page_{{ business_name }}_crud(
offset=offset,
limit=page_size,
order_by=order_by_list,
search=search_dict
)
return result
@classmethod
async def create_{{ business_name }}_service(cls, auth: AuthSchema, data: {{ class_name }}CreateSchema) -> dict:
"""创建"""
# 检查唯一性约束
{% for column in columns %}
{% if column.is_unique == '1' %}
obj = await {{ class_name }}CRUD(auth).get({{ column.column_name }}=data.{{ column.column_name }})
if obj:
raise CustomException(msg='创建失败,{{ column.column_comment }}已存在')
{% endif %}
{% endfor %}
obj = await {{ class_name }}CRUD(auth).create_{{ business_name }}_crud(data=data)
return {{ class_name }}OutSchema.model_validate(obj).model_dump()
@classmethod
async def update_{{ business_name }}_service(cls, auth: AuthSchema, id: int, data: {{ class_name }}UpdateSchema) -> dict:
"""更新"""
# 检查数据是否存在
obj = await {{ class_name }}CRUD(auth).get_by_id_{{ business_name }}_crud(id=id)
if not obj:
raise CustomException(msg='更新失败,该数据不存在')
# 检查唯一性约束
{% for column in columns %}
{% if column.is_unique == '1' %}
exist_obj = await {{ class_name }}CRUD(auth).get({{ column.column_name }}=data.{{ column.column_name }})
if exist_obj and exist_obj.id != id:
raise CustomException(msg='更新失败,{{ column.column_comment }}重复')
{% endif %}
{% endfor %}
obj = await {{ class_name }}CRUD(auth).update_{{ business_name }}_crud(id=id, data=data)
return {{ class_name }}OutSchema.model_validate(obj).model_dump()
@classmethod
async def delete_{{ business_name }}_service(cls, auth: AuthSchema, ids: list[int]) -> None:
"""删除"""
if len(ids) < 1:
raise CustomException(msg='删除失败,删除对象不能为空')
for id in ids:
obj = await {{ class_name }}CRUD(auth).get_by_id_{{ business_name }}_crud(id=id)
if not obj:
raise CustomException(msg=f'删除失败,ID为{id}的数据不存在')
await {{ class_name }}CRUD(auth).delete_{{ business_name }}_crud(ids=ids)
@classmethod
async def set_available_{{ business_name }}_service(cls, auth: AuthSchema, data: BatchSetAvailable) -> None:
"""批量设置状态"""
await {{ class_name }}CRUD(auth).set_available_{{ business_name }}_crud(ids=data.ids, status=data.status)
@classmethod
async def batch_export_{{ business_name }}_service(cls, obj_list: list[dict]) -> bytes:
"""批量导出"""
mapping_dict = {
{% for column in columns %}
'{{ column.column_name }}': '{{ column.column_comment }}',
{% endfor %}
'updated_id': '更新者ID',
}
data = obj_list.copy()
for item in data:
# 状态转换
if 'status' in item:
item['status'] = '启用' if item.get('status') == '0' else '停用'
# 创建者转换
creator_info = item.get('creator')
if isinstance(creator_info, dict):
item['creator'] = creator_info.get('name', '未知')
elif creator_info is None:
item['creator'] = '未知'
return ExcelUtil.export_list2excel(list_data=data, mapping_dict=mapping_dict)
@classmethod
async def batch_import_{{ business_name }}_service(cls, auth: AuthSchema, file: UploadFile, update_support: bool = False) -> str:
"""批量导入"""
header_dict = {
{% for column in columns %}
'{{ column.column_comment }}': '{{ column.column_name }}',
{% endfor %}
}
try:
contents = await file.read()
df = pd.read_excel(io.BytesIO(contents))
await file.close()
if df.empty:
raise CustomException(msg="导入文件为空")
missing_headers = [header for header in header_dict.keys() if header not in df.columns]
if missing_headers:
raise CustomException(msg=f"导入文件缺少必要的列: {', '.join(missing_headers)}")
df.rename(columns=header_dict, inplace=True)
# 验证必填字段
{% for column in columns %}
{% if column.required == '1' %}
errors = []
missing_rows = df[df['{{ column.column_name }}'].isnull()].index.tolist()
if missing_rows:
field_name = [k for k,v in header_dict.items() if v == field][0]
rows_str = "、".join([str(i+1) for i in missing_rows])
errors.append(f"{field_name}不能为空,第{rows_str}行")
if errors:
raise CustomException(msg=f"导入失败,以下行缺少必要字段:\n{'; '.join(errors)}")
{% endif %}
{% endfor %}
error_msgs = []
success_count = 0
count = 0
for index, row in df.iterrows():
count += 1
try:
data = {
{% for column in columns %}
"{{ column.column_name }}": row['{{ column.column_name }}'],
{% endfor %}
}
# 使用CreateSchema做校验后入库
create_schema = {{ class_name }}CreateSchema.model_validate(data)
# 检查唯一性约束
{% for column in columns %}
{% if column.is_unique == '1' %}
exists_obj = await {{ class_name }}CRUD(auth).get({{ column.column_name }}=create_schema.{{ column.column_name }})
if exists_obj:
if update_support:
await {{ class_name }}CRUD(auth).update(id=exists_obj.id, data=create_schema)
success_count += 1
else:
error_msgs.append(f"第{count}行: {{ column.column_comment }} {create_schema.{{ column.column_name }}} 已存在")
continue
{% endif %}
{% endfor %}
await {{ class_name }}CRUD(auth).create_{{ business_name }}_crud(data=create_schema)
success_count += 1
except Exception as e:
error_msgs.append(f"第{count}行: {str(e)}")
continue
result = f"成功导入 {success_count} 条数据"
if error_msgs:
result += "\n错误信息:\n" + "\n".join(error_msgs)
return result
except Exception as e:
log.error(f"批量导入失败: {str(e)}")
raise CustomException(msg=f"导入失败: {str(e)}")
@classmethod
async def import_template_download_{{ business_name }}_service(cls) -> bytes:
"""下载导入模板"""
header_list = [
{% for column in columns %}
'{{ column.column_comment }}',
{% endfor %}
]
selector_header_list = []
option_list = []
# 添加下拉选项
{% for column in columns %}
{% if column.html_type == 'select' and column.dict_type %}
selector_header_list.append('{{ column.column_comment }}')
option_list.append({'{{ column.column_comment }}': []})
{% endif %}
{% endfor %}
return ExcelUtil.get_excel_template(
header_list=header_list,
selector_header_list=selector_header_list,
option_list=option_list
)
@@ -0,0 +1,137 @@
import request from "@/utils/request";
const API_PATH = "/{{ package_name }}/{{ business_name|lower }}";
const {{ class_name }}API = {
// 列表查询
list{{ class_name }}(query: {{ class_name }}PageQuery) {
return request<ApiResponse<PageResult<{{ class_name }}Table[]>>>({
url: `${API_PATH}/list`,
method: "get",
params: query,
});
},
// 详情查询
detail{{ class_name }}(id: number) {
return request<ApiResponse<{{ class_name }}Table>>({
url: `${API_PATH}/detail/${id}`,
method: "get",
});
},
// 新增
create{{ class_name }}(body: {{ class_name }}Form) {
return request<ApiResponse>({
url: `${API_PATH}/create`,
method: "post",
data: body,
});
},
// 修改(带主键)
update{{ class_name }}(id: number, body: {{ class_name }}Form) {
return request<ApiResponse>({
url: `${API_PATH}/update/${id}`,
method: "put",
data: body,
});
},
// 删除(支持批量)
delete{{ class_name }}(ids: number[]) {
return request<ApiResponse>({
url: `${API_PATH}/delete`,
method: "delete",
data: ids,
});
},
// 批量启用/停用
batch{{ class_name }}(body: BatchType) {
return request<ApiResponse>({
url: `${API_PATH}/available/setting`,
method: "patch",
data: body,
});
},
// 导出
export{{ class_name }}(query: {{ class_name }}PageQuery) {
return request<Blob>({
url: `${API_PATH}/export`,
method: "post",
data: query,
responseType: "blob",
});
},
// 下载导入模板
downloadTemplate{{ class_name }}() {
return request<Blob>({
url: `${API_PATH}/download/template`,
method: "post",
responseType: "blob",
});
},
// 导入
import{{ class_name }}(body: FormData) {
return request<ApiResponse>({
url: `${API_PATH}/import`,
method: "post",
data: body,
headers: { "Content-Type": "multipart/form-data" },
});
},
};
export default {{ class_name }}API;
// ------------------------------
// TS 类型声明
// ------------------------------
// 列表查询参数
export interface {{ class_name }}PageQuery extends PageQuery {
{% for column in columns %}
{% if column.is_query and column.column != "BETWEEN" and column.column_name not in ['created_time', 'updated_time'] %}
{{ column.column_name }}?: {{
'string' if ('status' in (column.python_field|lower)) or (column.html_type == 'radio')
else 'number' if column.is_pk == '1'
else 'number' if column.column_name in ['created_id', 'updated_id']
else 'string'
}};
{% endif %}
{% endfor %}
created_time?: string[];
updated_time?: string[];
}
// 列表展示项
export interface {{ class_name }}Table extends BaseType {
{% for column in columns %}
{% if column.column_name not in ['id', 'uuid', 'status', 'description', 'created_time', 'updated_time'] %}
{{ column.column_name }}?: {{
'boolean' if ('status' in (column.column_name|lower)) or (column.html_type == 'radio')
else 'number' if column.is_pk == 1
else 'string'
}};
{% endif %}
{% endfor %}
created_by?: CommonType;
updated_by?: CommonType;
}
// 新增/修改/详情表单参数
export interface {{ class_name }}Form extends BaseFormType {
{% for column in columns %}
{% if (column.is_insert == 1 or column.is_edit == 1) and column.column_name not in ['id', 'uuid', 'status', 'description', 'created_time', 'updated_time', 'created_id', 'updated_id'] %}
{{ column.column_name }}?: {{
'boolean' if ('status' in (column.column_name|lower)) or (column.html_type == 'radio')
else 'number' if column.is_pk == 1
else 'string'
}};
{% endif %}
{% endfor %}
}
@@ -0,0 +1,884 @@
<!-- {{ function_name }} -->
<template>
<div class="app-container">
<!-- 搜索区域 -->
<div v-show="visible" class="search-container">
<el-form
ref="queryFormRef"
:model="queryFormData"
label-suffix=":"
:inline="true"
@submit.prevent="handleQuery"
>
{% for column in columns %}
{% if column.is_query == 1 %}
{% set dict_type = column.dict_type %}
{% set column_comment = column.column_comment if column.column_comment else '' %}
{% set parentheseIndex = column_comment.find("") %}
{% set comment = column_comment[:parentheseIndex] if parentheseIndex != -1 else column_comment %}
{% if column.column_name == "status" %}
<el-form-item prop="status" label="状态">
<el-select
v-model="queryFormData.status"
placeholder="请选择状态"
style="width: 170px"
clearable
>
<el-option value="0" label="启用" />
<el-option value="1" label="停用" />
</el-select>
</el-form-item>
{% elif column.column_name == "created_id"%}
<el-form-item v-if="isExpand" prop="created_id" label="创建人">
<UserTableSelect
v-model="queryFormData.created_id"
@confirm-click="handleConfirm"
@clear-click="handleQuery"
/>
</el-form-item>
{% elif column.column_name == "updated_id"%}
<el-form-item v-if="isExpand" prop="updated_id" label="更新人">
<UserTableSelect
v-model="queryFormData.updated_id"
@confirm-click="handleConfirm"
@clear-click="handleQuery"
/>
</el-form-item>
{% elif column.column_name == "created_time"%}
<el-form-item v-if="isExpand" prop="created_time" label="创建时间">
<DatePicker
v-model="createdDateRange"
@update:model-value="handleCreatedDateRangeChange"
/>
</el-form-item>
{% elif column.column_name == "updated_time"%}
<el-form-item v-if="isExpand" prop="updated_time" label="更新时间">
<DatePicker
v-model="updatedDateRange"
@update:model-value="handleUpdatedDateRangeChange"
/>
</el-form-item>
{% elif column.html_type == "input" %}
<el-form-item label="{{ comment }}" prop="{{ column.column_name }}">
<el-input v-model="queryFormData.{{ column.column_name }}" placeholder="请输入{{ comment }}" clearable />
</el-form-item>
{% elif (column.html_type == "select" or column.html_type == "radio") and dict_type != "" %}
<el-form-item label="{{ comment }}" prop="{{ column.column_name }}">
<el-select v-model="queryFormData.{{ column.column_name }}" placeholder="请选择{{ comment }}" style="width: 180px" clearable>
<el-option v-for="dict in dictStore.getDictArray('{{ dict_type }}')" :key="dict.dict_value" :label="dict.dict_label" :value="dict.dict_value" />
</el-select>
</el-form-item>
{% elif (column.html_type == "select" or column.html_type == "radio") and dict_type %}
<el-form-item label="{{ comment }}" prop="{{ column.column_name }}">
<el-select v-model="queryFormData.{{ column.column_name }}" placeholder="请选择{{ comment }}" clearable>
<el-option label="请选择字典生成" value="" />
</el-select>
</el-form-item>
{% elif column.html_type == "datetime" and column.query_type != "BETWEEN" %}
<el-form-item label="{{ comment }}" prop="{{ column.column_name }}">
<el-date-picker v-model="queryFormData.{{ column.column_name }}" type="date" value-format="YYYY-MM-DD" clearable placeholder="请选择{{ comment }}" />
</el-form-item>
{% endif %}
{% endif %}
{% endfor %}
<!-- 查询、重置、展开/收起按钮 -->
<el-form-item>
<el-button
v-hasPerm="['{{ module_name }}:{{ business_name }}:query']"
type="primary"
icon="search"
@click="handleQuery"
>
查询
</el-button>
<el-button
v-hasPerm="['{{ module_name }}:{{ business_name }}:query']"
icon="refresh"
@click="handleResetQuery"
>
重置
</el-button>
<!-- 展开/收起 -->
<template v-if="isExpandable">
<el-link class="ml-3" type="primary" underline="never" @click="isExpand = !isExpand">
{{ '{{' }} isExpand ? "收起" : "展开" {{ '}}' }}
<el-icon>
<template v-if="isExpand">
<ArrowUp />
</template>
<template v-else>
<ArrowDown />
</template>
</el-icon>
</el-link>
</template>
</el-form-item>
</el-form>
</div>
<!-- 内容区域 -->
<el-card class="data-table">
<template #header>
<div class="card-header">
<span>
{{ function_name }}列表
<el-tooltip content="{{ function_name }}列表">
<QuestionFilled class="w-4 h-4 mx-1" />
</el-tooltip>
</span>
</div>
</template>
<!-- 功能区域 -->
<div class="data-table__toolbar">
<div class="data-table__toolbar--left">
<el-row :gutter="10">
<el-col :span="1.5">
<el-button
v-hasPerm="['{{ module_name }}:{{ business_name }}:create']"
type="success"
icon="plus"
@click="handleOpenDialog('create')"
>
新增
</el-button>
</el-col>
<el-col :span="1.5">
<el-button
v-hasPerm="['{{ module_name }}:{{ business_name }}:delete']"
type="danger"
icon="delete"
:disabled="selectIds.length === 0"
@click="handleDelete(selectIds)"
>
批量删除
</el-button>
</el-col>
<el-col :span="1.5">
<el-dropdown v-hasPerm="['{{ module_name }}:{{ business_name }}:batch']" trigger="click">
<el-button type="default" :disabled="selectIds.length === 0" icon="ArrowDown">
更多
</el-button>
<template #dropdown>
<el-dropdown-menu>
<el-dropdown-item :icon="Check" @click="handleMoreClick('0')">
批量启用
</el-dropdown-item>
<el-dropdown-item :icon="CircleClose" @click="handleMoreClick('1')">
批量停用
</el-dropdown-item>
</el-dropdown-menu>
</template>
</el-dropdown>
</el-col>
</el-row>
</div>
<div class="data-table__toolbar--right">
<el-row :gutter="10">
<el-col :span="1.5">
<el-tooltip content="导入">
<el-button
v-hasPerm="['{{ module_name }}:{{ business_name }}:import']"
type="success"
icon="upload"
circle
@click="handleOpenImportDialog"
/>
</el-tooltip>
</el-col>
<el-col :span="1.5">
<el-tooltip content="导出">
<el-button
v-hasPerm="['{{ module_name }}:{{ business_name }}:export']"
type="warning"
icon="download"
circle
@click="handleOpenExportsModal"
/>
</el-tooltip>
</el-col>
<el-col :span="1.5">
<el-tooltip content="搜索显示/隐藏">
<el-button
v-hasPerm="['*:*:*']"
type="info"
icon="search"
circle
@click="visible = !visible"
/>
</el-tooltip>
</el-col>
<el-col :span="1.5">
<el-tooltip content="刷新">
<el-button
v-hasPerm="['{{ module_name }}:{{ business_name }}:query']"
type="primary"
icon="refresh"
circle
@click="handleRefresh"
/>
</el-tooltip>
</el-col>
<el-col :span="1.5">
<el-popover placement="bottom" trigger="click">
<template #reference>
<el-button type="danger" icon="operation" circle></el-button>
</template>
<el-scrollbar max-height="350px">
<template v-for="column in tableColumns" :key="column.prop">
<el-checkbox v-if="column.prop" v-model="column.show" :label="column.label" />
</template>
</el-scrollbar>
</el-popover>
</el-col>
</el-row>
</div>
</div>
<!-- 表格区域:系统配置列表 -->
<el-table
ref="tableRef"
v-loading="loading"
:data="pageTableData"
highlight-current-row
class="data-table__content"
:height="450"
border
stripe
@selection-change="handleSelectionChange"
>
<template #empty>
<el-empty :image-size="80" description="暂无数据" />
</template>
<el-table-column
v-if="tableColumns.find((col) => col.prop === 'selection')?.show"
type="selection"
min-width="55"
align="center"
/>
<el-table-column
v-if="tableColumns.find((col) => col.prop === 'index')?.show"
fixed
label="序号"
min-width="60"
>
<template #default="scope">
{{ '{{' }} (queryFormData.page_no - 1) * queryFormData.page_size + scope.$index + 1 {{ '}}' }}
</template>
</el-table-column>
{% for column in columns %}
{% set python_field = column.column_name %}
{% set column_comment = column.column_comment if column.column_comment else '' %}
{% set parentheseIndex = column_comment.find("") %}
{% set comment = column_comment[:parentheseIndex] if parentheseIndex != -1 else column_comment %}
{% if column.is_list == 1 %}
<el-table-column
v-if="tableColumns.find((col) => col.prop === '{{ python_field }}')?.show"
label="{{ comment }}"
prop="{{ python_field }}"
min-width="140"
/>
{% if python_field in ['status', 'created_id', 'updated_id'] %}
<el-table-column
v-if="tableColumns.find((col) => col.prop === '{{ python_field }}')?.show"
label="{{ comment }}"
prop="{{ python_field }}"
min-width="140"
>
{% if python_field == "status" %}
<template #default="scope">
<el-tag :type="scope.row.status == '0' ? 'success' : 'info'">
{{ '{{' }} scope.row.status == "0" ? "启用" : "停用" {{ '}}' }}
</el-tag>
</template>
{% elif python_field == "created_id" %}
<template #default="scope">
<el-tag>{{ '{{' }} scope.row.created_by?.name {{ '}}' }}</el-tag>
</template>
{% elif python_field == "updated_id" %}
<template #default="scope">
<el-tag>{{ '{{' }} scope.row.updated_by?.name {{ '}}' }}</el-tag>
</template>
{% endif %}
</el-table-column>
{% endif %}
{% endif %}
{% endfor %}
<el-table-column
v-if="tableColumns.find((col) => col.prop === 'operation')?.show"
fixed="right"
label="操作"
align="center"
min-width="180"
>
<template #default="scope">
<el-button
v-hasPerm="['{{ module_name }}:{{ business_name }}:detail']"
type="info"
size="small"
link
icon="document"
@click="handleOpenDialog('detail', scope.row.id)"
>
详情
</el-button>
<el-button
v-hasPerm="['{{ module_name }}:{{ business_name }}:update']"
type="primary"
size="small"
link
icon="edit"
@click="handleOpenDialog('update', scope.row.id)"
>
编辑
</el-button>
<el-button
v-hasPerm="['{{ module_name }}:{{ business_name }}:delete']"
type="danger"
size="small"
link
icon="delete"
@click="handleDelete([scope.row.id])"
>
删除
</el-button>
</template>
</el-table-column>
</el-table>
<!-- 分页区域 -->
<template #footer>
<pagination
v-model:total="total"
v-model:page="queryFormData.page_no"
v-model:limit="queryFormData.page_size"
@pagination="loadingData"
/>
</template>
</el-card>
<!-- 弹窗区域 -->
<el-dialog
v-model="dialogVisible.visible"
:title="dialogVisible.title"
@close="handleCloseDialog"
>
<!-- 详情 -->
<template v-if="dialogVisible.type === 'detail'">
<el-descriptions :column="4" border>
{% for column in columns %}
{% set column_comment = column.column_comment if column.column_comment else '' %}
{% set parentheseIndex = column_comment.find("") %}
{% set comment = column_comment[:parentheseIndex] if parentheseIndex != -1 else column_comment %}
{% if column.column_name == 'status' %}
<el-descriptions-item label="状态" :span="2">
<el-tag :type="detailFormData.status == '0' ? 'success' : 'danger'">
{{ '{{' }} detailFormData.status == "0" ? "启用" : "停用" {{ '}}' }}
</el-tag>
</el-descriptions-item>
{% elif column.column_name == 'created_id' %}
<el-descriptions-item label="创建人" :span="2">
{{ '{{' }} detailFormData.created_by?.name {{ '}}' }}
</el-descriptions-item>
{% elif column.column_name == 'updated_id' %}
<el-descriptions-item label="更新人" :span="2">
{{ '{{' }} detailFormData.updated_by?.name {{ '}}' }}
</el-descriptions-item>
{% else %}
<el-descriptions-item label="{{ comment }}" :span="2">
{{ '{{' }} detailFormData.{{ column.column_name }} {{ '}}' }}
</el-descriptions-item>
{% endif %}
{% endfor %}
</el-descriptions>
</template>
<!-- 新增、编辑表单 -->
<template v-else>
<el-form
ref="dataFormRef"
:model="formData"
:rules="rules"
label-suffix=":"
label-width="auto"
label-position="right"
>
{% for column in columns %}
{% if column.is_insert == 1 or column.is_edit == 1 %}
{% set dict_type = column.dict_type %}
{% set column_comment = column.column_comment if column.column_comment else '' %}
{% set parentheseIndex = column_comment.find("") %}
{% set comment = column_comment[:parentheseIndex] if parentheseIndex != -1 else column_comment %}
{% set required = 'true' if column.is_nullable == '1' else 'false' %}
{% if column.column_name not in ['id', 'uuid', 'created_time', 'updated_time', 'created_id', 'updated_id'] %}
{% if column.column_name == "status" %}
<el-form-item label="状态" prop="status" :required="true">
<el-radio-group v-model="formData.status">
<el-radio value="0">启用</el-radio>
<el-radio value="1">停用</el-radio>
</el-radio-group>
</el-form-item>
{% elif column.column_name == "description" %}
<el-form-item label="描述" prop="description">
<el-input
v-model="formData.description"
:rows="4"
:maxlength="100"
show-word-limit
type="textarea"
placeholder="请输入描述"
/>
</el-form-item>
{% elif column.html_type == "input" %}
<el-form-item label="{{ comment }}" prop="{{ column.column_name }}" :required="{{ required }}">
<el-input v-model="formData.{{ column.column_name }}" placeholder="请输入{{ comment }}" />
</el-form-item>
{% elif column.html_type == "textarea" %}
<el-form-item label="{{ comment }}" prop="{{ column.column_name }}" :required="{{ required }}">
<el-input v-model="formData.{{ column.column_name }}" type="textarea" placeholder="请输入{{ comment }}" rows="4" :maxlength="100" show-word-limit />
</el-form-item>
{% elif (column.html_type == "select" or column.html_type == "radio") and dict_type != "" %}
<el-form-item label="{{ comment }}" prop="{{ column.column_name }}" :required="{{ required }}">
<el-select v-model="formData.{{ column.column_name }}" placeholder="请选择{{ comment }}">
<el-option v-for="dict in dictStore.getDictArray('{{ dict_type }}')" :key="dict.dict_value" :label="dict.dict_label" :value="dict.dict_value" />
</el-select>
</el-form-item>
{% elif (column.html_type == "select" or column.html_type == "radio") and dict_type %}
<el-form-item label="{{ comment }}" prop="{{ column.column_name }}" :required="{{ required }}">
<el-select v-model="formData.{{ column.column_name }}" placeholder="请选择{{ comment }}">
<el-option label="请选择字典生成" value="" />
</el-select>
</el-form-item>
{% elif column.html_type == "date" %}
<el-form-item label="{{ comment }}" prop="{{ column.column_name }}" :required="{{ required }}">
<el-date-picker v-model="formData.{{ column.column_name }}" type="date" value-format="YYYY-MM-DD" placeholder="请选择{{ comment }}" />
</el-form-item>
{% elif column.html_type == "datetime" %}
<el-form-item label="{{ comment }}" prop="{{ column.column_name }}" :required="{{ required }}">
<el-date-picker v-model="formData.{{ column.column_name }}" type="datetime" value-format="YYYY-MM-DD HH:mm:ss" placeholder="请选择{{ comment }}" />
</el-form-item>
{% elif column.html_type == "checkbox" %}
<el-form-item label="{{ comment }}" prop="{{ column.column_name }}">
<el-checkbox v-model="formData.{{ column.column_name }}">{{ comment }}</el-checkbox>
</el-form-item>
{% elif column.html_type == "imageUpload" %}
<el-form-item label="{{ comment }}" prop="{{ column.column_name }}">
<SingleImageUpload v-model="formData.{{ column.column_name }}" />
</el-form-item>
{% endif %}
{% endif %}
{% endif %}
{% endfor %}
</el-form>
</template>
<template #footer>
<div class="dialog-footer">
<!-- 详情弹窗不需要确定按钮的提交逻辑 -->
<el-button @click="handleCloseDialog">取消</el-button>
<el-button v-if="dialogVisible.type !== 'detail'" type="primary" @click="handleSubmit">
确定
</el-button>
<el-button v-else type="primary" @click="handleCloseDialog">确定</el-button>
</div>
</template>
</el-dialog>
<!-- 导入弹窗 -->
<ImportModal
v-model="importDialogVisible"
:content-config="curdContentConfig"
@upload="handleUpload"
/>
<!-- 导出弹窗 -->
<ExportModal
v-model="exportsDialogVisible"
:content-config="curdContentConfig"
:query-params="queryFormData"
:page-data="pageTableData"
:selection-data="selectionRows"
/>
</div>
</template>
<script setup lang="ts">
defineOptions({
name: "{{ class_name }}",
inheritAttrs: false,
});
import { ref, reactive, onMounted } from "vue";
import { ElMessage, ElMessageBox } from "element-plus";
import { QuestionFilled, ArrowUp, ArrowDown, Check, CircleClose } from "@element-plus/icons-vue";
import { formatToDateTime } from "@/utils/dateUtil";
import { useDictStore } from "@/store";
import { ResultEnum } from "@/enums/api/result.enum";
import DatePicker from "@/components/DatePicker/index.vue";
import type { IContentConfig } from "@/components/CURD/types";
import ImportModal from "@/components/CURD/ImportModal.vue";
import ExportModal from "@/components/CURD/ExportModal.vue";
import {{ class_name }}API, {
{{ class_name }}PageQuery,
{{ class_name }}Table,
{{ class_name }}Form,
} from "@/api/{{ module_name }}/{{ business_name }}";
const visible = ref(true);
const isExpand = ref(false);
const isExpandable = ref(true);
const queryFormRef = ref();
const dataFormRef = ref();
const total = ref(0);
const selectIds = ref<number[]>([]);
const selectionRows = ref<{{ class_name }}Table[]>([]);
const loading = ref(false);
// 字典仓库与需要加载的字典类型
const dictStore = useDictStore();
const dictTypes: any = [
{% for column in columns %}
{% if column.dict_type %}
"{{ column.dict_type }}",
{% endif %}
{% endfor %}
];
// 分页表单
const pageTableData = ref<{{ class_name }}Table[]>([]);
// 表格列配置
const tableColumns = ref([
{ prop: "selection", label: "选择框", show: true },
{ prop: "index", label: "序号", show: true },
{% for column in columns %}
{% if column.is_list == 1 %}
{ prop: "{{ column.column_name }}", label: "{{ column.column_comment or column.column_name }}", show: true },
{% endif %}
{% endfor %}
{ prop: "operation", label: "操作", show: true },
]);
// 导出列(不含选择/序号/操作)
const exportColumns = [
{% for column in columns %}
{% if column.is_list == 1 %}
{ prop: "{{ column.column_name }}", label: "{{ column.column_comment or column.column_name }}" },
{% endif %}
{% endfor %}
];
// 导入/导出配置
const curdContentConfig = {
permPrefix: "{{ module_name }}:{{ business_name }}",
cols: exportColumns as any,
importTemplate: () => {{ class_name }}API.downloadTemplate{{ class_name }}(),
exportsAction: async (params: any) => {
const query: any = { ...params };
query.status = "0";
query.page_no = 1;
query.page_size = 9999;
const all: any[] = [];
while (true) {
const res = await {{ class_name }}API.list{{ class_name }}(query);
const items = res.data?.data?.items || [];
const total = res.data?.data?.total || 0;
all.push(...items);
if (all.length >= total || items.length === 0) break;
query.page_no += 1;
}
return all;
},
} as unknown as IContentConfig;
// 详情表单
const detailFormData = ref<{{ class_name }}Table>({});
// 日期范围临时变量
const createdDateRange = ref<[Date, Date] | []>([]);
// 更新时间范围临时变量
const updatedDateRange = ref<[Date, Date] | []>([]);
// 处理创建时间范围变化
function handleCreatedDateRangeChange(range: [Date, Date]) {
createdDateRange.value = range;
if (range && range.length === 2) {
queryFormData.created_time = [formatToDateTime(range[0]), formatToDateTime(range[1])];
} else {
queryFormData.created_time = undefined;
}
}
// 处理更新时间范围变化
function handleUpdatedDateRangeChange(range: [Date, Date]) {
updatedDateRange.value = range;
if (range && range.length === 2) {
queryFormData.updated_time = [formatToDateTime(range[0]), formatToDateTime(range[1])];
} else {
queryFormData.updated_time = undefined;
}
}
// 分页查询参数
const queryFormData = reactive<{{ class_name }}PageQuery>({
page_no: 1,
page_size: 10,
{% for column in columns %}
{% if column.is_query == 1 %}
{{ column.column_name }}: undefined,
{% endif %}
{% endfor %}
});
// 编辑表单
const formData = reactive<{{ class_name }}Form>({
{% for column in columns %}
{% if column.is_insert == 1 or column.is_edit == 1 %}
{% if column.column_name not in ['uuid', 'created_time', 'updated_time', 'created_id', 'updated_id'] %}
{{ column.column_name }}: undefined,
{% endif %}
{% endif %}
{% endfor %}
});
// 弹窗状态
const dialogVisible = reactive({
title: "",
visible: false,
type: "create" as "create" | "update" | "detail",
});
// 表单验证规则
const rules = reactive({
{% for column in columns %}
{% if column.is_insert == 1 or column.is_edit == 1 %}
{% set required = 'true' if column.is_nullable == 1 else 'false' %}
{{ column.column_name }}: [{ required: {{ required }}, message: "请输入{{ column.column_comment or column.column_name }}", trigger: "blur" }],
{% endif %}
{% endfor %}
});
// 导入弹窗显示状态
const importDialogVisible = ref(false);
// 导出弹窗显示状态
const exportsDialogVisible = ref(false);
// 打开导入弹窗
function handleOpenImportDialog() {
importDialogVisible.value = true;
}
// 打开导出弹窗
function handleOpenExportsModal() {
exportsDialogVisible.value = true;
}
// 列表刷新
async function handleRefresh() {
await loadingData();
}
// 加载表格数据
async function loadingData() {
loading.value = true;
try {
const response = await {{ class_name }}API.list{{ class_name }}(queryFormData);
pageTableData.value = response.data.data.items;
total.value = response.data.data.total;
} catch (error: any) {
console.error(error);
} finally {
loading.value = false;
}
}
// 查询(重置页码后获取数据)
async function handleQuery() {
queryFormData.page_no = 1;
loadingData();
}
// 选择创建人后触发查询
function handleConfirm() {
handleQuery();
}
// 重置查询
async function handleResetQuery() {
queryFormRef.value.resetFields();
queryFormData.page_no = 1;
// 重置日期范围选择器
createdDateRange.value = [];
updatedDateRange.value = [];
queryFormData.created_time = undefined;
queryFormData.updated_time = undefined;
loadingData();
}
// 定义初始表单数据常量
const initialFormData: {{ class_name }}Form = {
{% for column in columns %}
{% if column.is_insert == 1 or column.is_edit == 1 %}
{% if column.column_name not in ['uuid', 'created_time', 'updated_time', 'created_id', 'updated_id'] %}
{{ column.column_name }}: undefined,
{% endif %}
{% endif %}
{% endfor %}
};
// 重置表单
async function resetForm() {
if (dataFormRef.value) {
dataFormRef.value.resetFields();
dataFormRef.value.clearValidate();
}
// 完全重置 formData 为初始状态
Object.assign(formData, initialFormData);
}
// 行复选框选中项变化
async function handleSelectionChange(selection: any) {
selectIds.value = selection.map((item: any) => item.id);
selectionRows.value = selection;
}
// 关闭弹窗
async function handleCloseDialog() {
dialogVisible.visible = false;
resetForm();
}
// 打开弹窗
async function handleOpenDialog(type: "create" | "update" | "detail", id?: number) {
dialogVisible.type = type;
if (id) {
const response = await {{ class_name }}API.detail{{ class_name }}(id);
if (type === "detail") {
dialogVisible.title = "详情";
Object.assign(detailFormData.value, response.data.data);
} else if (type === "update") {
dialogVisible.title = "修改";
Object.assign(formData, response.data.data);
}
} else {
dialogVisible.title = "新增{{ class_name }}";
{% for column in columns %}
{% if column.is_insert == 1 or column.is_edit == 1 %}
{% if column.column_name not in ['uuid', 'created_time', 'updated_time', 'created_id', 'updated_id'] %}
formData.{{ column.column_name }} = undefined;
{% endif %}
{% endif %}
{% endfor %}
}
dialogVisible.visible = true;
}
// 提交表单(防抖)
async function handleSubmit() {
// 表单校验
dataFormRef.value.validate(async (valid: any) => {
if (valid) {
loading.value = true;
// 根据弹窗传入的参数(deatil\create\update)判断走什么逻辑
const id = formData.id;
if (id) {
try {
await {{ class_name }}API.update{{ class_name }}(id, { id, ...formData });
dialogVisible.visible = false;
resetForm();
handleCloseDialog();
handleResetQuery();
} catch (error: any) {
console.error(error);
} finally {
loading.value = false;
}
} else {
try {
await {{ class_name }}API.create{{ class_name }}(formData);
dialogVisible.visible = false;
resetForm();
handleCloseDialog();
handleResetQuery();
} catch (error: any) {
console.error(error);
} finally {
loading.value = false;
}
}
}
});
}
// 删除、批量删除
async function handleDelete(ids: number[]) {
ElMessageBox.confirm("确认删除该项数据?", "警告", {
confirmButtonText: "确定",
cancelButtonText: "取消",
type: "warning",
})
.then(async () => {
try {
loading.value = true;
await {{ class_name }}API.delete{{ class_name }}(ids);
handleResetQuery();
} catch (error: any) {
console.error(error);
} finally {
loading.value = false;
}
})
.catch(() => {
ElMessageBox.close();
});
}
// 批量启用/停用
async function handleMoreClick(status: string) {
if (selectIds.value.length) {
ElMessageBox.confirm(`确认${status === "0" ? "启用" : "停用"}该项数据?`, "警告", {
confirmButtonText: "确定",
cancelButtonText: "取消",
type: "warning",
})
.then(async () => {
try {
loading.value = true;
await {{ class_name }}API.batch{{ class_name }}({ ids: selectIds.value, status });
handleResetQuery();
} catch (error: any) {
console.error(error);
} finally {
loading.value = false;
}
})
.catch(() => {
ElMessageBox.close();
});
}
}
// 处理上传
const handleUpload = async (formData: FormData) => {
try {
const response = await {{ class_name }}API.import{{ class_name }}(formData);
if (response.data.code === ResultEnum.SUCCESS) {
ElMessage.success(`${response.data.msg}${response.data.data}`);
importDialogVisible.value = false;
await handleQuery();
}
} catch (error: any) {
console.error(error);
}
};
onMounted(async () => {
// 预加载字典数据
if (dictTypes.length > 0) {
await dictStore.getDict(dictTypes);
}
loadingData();
});
</script>
<style lang="scss" scoped></style>
@@ -0,0 +1,2 @@
# -*- coding: utf-8 -*-
@@ -0,0 +1,241 @@
# -*- coding: utf-8 -*-
import re
from app.common.constant import GenConstant
from app.utils.string_util import StringUtil
from app.plugin.module_generator.gencode.schema import GenTableOutSchema, GenTableSchema, GenTableColumnSchema
class GenUtils:
"""代码生成器工具类"""
@classmethod
def init_table(cls, gen_table: GenTableSchema) -> None:
"""
初始化表信息
参数:
- gen_table (GenTableSchema): 业务表对象
返回:
- None
"""
# 只有当字段为None时才设置默认值
gen_table.class_name = cls.convert_class_name(gen_table.table_name or "")
gen_table.package_name = 'gencode'
gen_table.module_name = f'module_{gen_table.package_name}'
gen_table.business_name = gen_table.table_name
gen_table.function_name = re.sub(r'(?:表|测试)', '', gen_table.table_comment or "")
@classmethod
def init_column_field(cls, column: GenTableColumnSchema, table: GenTableOutSchema) -> None:
"""
初始化列属性字段
参数:
- column (GenTableColumnSchema): 业务表字段对象
- table (GenTableOutSchema): 业务表对象
返回:
- None
"""
data_type = cls.get_db_type(column.column_type or "")
column_name = column.column_name or ""
if not table.id:
raise ValueError("业务表ID不能为空")
column.table_id = table.id
column.python_field = cls.to_camel_case(column_name)
# 只有当python_type为None时才设置默认类型
column.python_type = StringUtil.get_mapping_value_by_key_ignore_case(GenConstant.DB_TO_PYTHON, data_type)
if column.column_length is None:
column.column_length = ''
if column.column_default is None:
column.column_default = ''
if column.html_type is None:
if cls.arrays_contains(GenConstant.COLUMNTYPE_STR, data_type) or cls.arrays_contains(
GenConstant.COLUMNTYPE_TEXT, data_type
):
# 字符串长度超过500设置为文本域
column_length = cls.get_column_length(column.column_type or "")
html_type = (
GenConstant.HTML_TEXTAREA
if column_length >= 500 or cls.arrays_contains(GenConstant.COLUMNTYPE_TEXT, data_type)
else GenConstant.HTML_INPUT
)
column.html_type = html_type
elif cls.arrays_contains(GenConstant.COLUMNTYPE_TIME, data_type):
column.html_type = GenConstant.HTML_DATETIME
elif cls.arrays_contains(GenConstant.COLUMNTYPE_NUMBER, data_type):
column.html_type = GenConstant.HTML_INPUT
elif column_name.lower().endswith("status"):
column.html_type = GenConstant.HTML_RADIO
elif column_name.lower().endswith("type") or column_name.lower().endswith("sex"):
column.html_type = GenConstant.HTML_SELECT
elif column_name.lower().endswith("image"):
column.html_type = GenConstant.HTML_IMAGE_UPLOAD
elif column_name.lower().endswith("file"):
column.html_type = GenConstant.HTML_FILE_UPLOAD
elif column_name.lower().endswith("content"):
column.html_type = GenConstant.HTML_EDITOR
else:
column.html_type = GenConstant.HTML_INPUT
# 只有当is_insert为None时才设置插入字段(默认所有字段都需要插入)
if column.is_insert:
column.is_insert = GenConstant.REQUIRE
else:
column.is_insert = False
# 只有当is_edit为None时才设置编辑字段
if not cls.arrays_contains(GenConstant.COLUMNNAME_NOT_EDIT, column_name) and not column.is_pk:
column.is_edit = GenConstant.REQUIRE
else:
column.is_edit = False
# 只有当is_list为None时才设置列表字段
if not cls.arrays_contains(GenConstant.COLUMNNAME_NOT_LIST, column_name) and not column.is_pk:
column.is_list = GenConstant.REQUIRE
else:
column.is_list = False
# 只有当is_query为None时才设置查询字段
if not cls.arrays_contains(GenConstant.COLUMNNAME_NOT_QUERY, column_name) and not column.is_pk:
column.is_query = GenConstant.REQUIRE
# 直接设置查询类型,因为我们已经确定这是一个查询字段
if column_name.lower().endswith('name') or data_type in ['varchar', 'char', 'text']:
column.query_type = GenConstant.QUERY_LIKE
else:
column.query_type = GenConstant.QUERY_EQ
else:
column.is_query = False
column.query_type = None
@classmethod
def arrays_contains(cls, arr, target_value) -> bool:
"""
检查目标值是否在数组中
注意从根本上解决问题现在确保传入的参数都是正确的类型
- arr 是列表类型且在GenConstant中定义
- target_value 不会是None
参数:
- arr: 数组类型
- target_value: 目标值
返回:
- bool: 如果目标值在数组中返回True否则返回False
"""
# 从根本上解决问题,不再需要复杂的防御性检查
# 因为现在我们确保传入的arr是GenConstant中定义的列表常量
# 并且target_value在调用前已经被处理过不会是None
# 对于包含括号的类型(如TINYINT(1)),需要特殊处理
# 先获取基本类型名称(不含括号)用于比较
target_str = str(target_value).lower()
target_base_type = target_str.split('(')[0] if '(' in target_str else target_str
for item in arr:
item_str = str(item).lower()
item_base_type = item_str.split('(')[0] if '(' in item_str else item_str
if target_base_type == item_base_type:
return True
return False
@classmethod
def convert_class_name(cls, table_name: str) -> str:
"""
表名转换成 Python 类名
参数:
- table_name (str): 业务表名
返回:
- str: Python 类名
"""
return StringUtil.convert_to_camel_case(table_name)
@classmethod
def replace_first(cls, input_string: str, search_list: list[str]) -> str:
"""
批量替换前缀
参数:
- input_string (str): 需要被替换的字符串
- search_list (list[str]): 可替换的字符串列表
返回:
- str: 替换后的字符串
"""
for search_string in search_list:
if input_string.startswith(search_string):
return input_string.replace(search_string, '', 1)
return input_string
@classmethod
def get_db_type(cls, column_type: str) -> str:
"""
获取数据库类型字段
参数:
- column_type (str): 字段类型
返回:
- str: 数据库类型
"""
# 特殊处理tinyint(1),保留括号和长度信息以便识别为布尔类型
if column_type.lower().startswith('tinyint(1)'):
return column_type
if '(' in column_type:
return column_type.split('(')[0]
return column_type
@classmethod
def get_column_length(cls, column_type: str) -> int:
"""
获取字段长度
参数:
- column_type (str): 字段类型例如 'varchar(255)' 'decimal(10,2)'
返回:
- int: 字段长度优先取第一个长度值无法解析时返回0
"""
if '(' in column_type:
length = len(column_type.split('(')[1].split(')')[0])
return length
return 0
@classmethod
def split_column_type(cls, column_type: str) -> list[str]:
"""
拆分列类型
参数:
- column_type (str): 字段类型
返回:
- list[str]: 拆分结果
"""
if '(' in column_type and ')' in column_type:
return column_type.split('(')[1].split(')')[0].split(',')
return []
@classmethod
def to_camel_case(cls, text: str) -> str:
"""
将字符串转换为驼峰命名
参数:
- text (str): 需要转换的字符串
返回:
- str: 驼峰命名
"""
parts = text.split('_')
return parts[0] + ''.join(word.capitalize() for word in parts[1:])
@@ -0,0 +1,441 @@
# -*- coding: utf-8 -*-
from datetime import datetime
from jinja2.environment import Environment
from jinja2 import Environment, FileSystemLoader, Template
from typing import Any
from app.common.constant import GenConstant
from app.config.path_conf import TEMPLATE_DIR
from app.config.setting import settings
from app.utils.common_util import CamelCaseUtil, SnakeCaseUtil
from app.utils.string_util import StringUtil
from app.plugin.module_generator.gencode.schema import GenTableOutSchema, GenTableColumnOutSchema
class Jinja2TemplateUtil:
"""
模板处理工具类
"""
# 项目路径
FRONTEND_PROJECT_PATH = 'frontend'
BACKEND_PROJECT_PATH = 'backend'
# 环境对象
_env = None
@classmethod
def get_env(cls):
"""
获取模板环境对象
参数:
-
返回:
- Environment: Jinja2 环境对象
"""
try:
if cls._env is None:
cls._env = Environment(
loader=FileSystemLoader(TEMPLATE_DIR),
autoescape=False, # 自动转义HTML
trim_blocks=True, # 删除多余的空行
lstrip_blocks=True, # 删除行首空格
keep_trailing_newline=True, # 保留行尾换行符
enable_async=True, # 开启异步支持
)
cls._env.filters.update(
{
'camel_to_snake': SnakeCaseUtil.camel_to_snake,
'snake_to_camel': CamelCaseUtil.snake_to_camel,
'get_sqlalchemy_type': cls.get_sqlalchemy_type
}
)
return cls._env
except Exception as e:
raise RuntimeError(f'初始化Jinja2模板引擎失败: {e}')
@classmethod
def get_template(cls, template_path: str) -> Template:
"""
获取模板
参数:
- template_path (str): 模板路径
返回:
- Template: Jinja2 模板对象
异常:
- TemplateNotFound: 模板未找到时抛出
"""
return cls.get_env().get_template(template_path)
@classmethod
def prepare_context(cls, gen_table: GenTableOutSchema) -> dict[str, Any]:
"""
准备模板变量
参数:
- gen_table (GenTableOutSchema): 生成表的配置信息
返回:
- Dict[str, Any]: 模板上下文字典
"""
# 处理options为None的情况
# if not gen_table.options:
# raise ValueError('请先完善生成配置信息')
class_name = gen_table.class_name or ''
module_name = gen_table.module_name or ''
business_name = gen_table.business_name or ''
package_name = gen_table.package_name or ''
function_name = gen_table.function_name or ''
context = {
'table_name': gen_table.table_name or '',
'table_comment': gen_table.table_comment or '',
'function_name': function_name if StringUtil.is_not_empty(function_name) else '【请填写功能名称】',
'class_name': class_name,
'module_name': module_name,
'business_name': business_name,
'base_package': cls.get_package_prefix(package_name),
'package_name': package_name,
'datetime': datetime.now().strftime('%Y-%m-%d %H:%M:%S'),
'pk_column': gen_table.pk_column,
'model_import_list': cls.get_model_import_list(gen_table),
'schema_import_list': cls.get_schema_import_list(gen_table),
'permission_prefix': cls.get_permission_prefix(module_name, business_name),
'columns': gen_table.columns or [],
'table': gen_table,
'dicts': cls.get_dicts(gen_table),
'db_type': settings.DATABASE_TYPE,
'column_not_add_show': GenConstant.COLUMNNAME_NOT_ADD_SHOW,
'column_not_edit_show': GenConstant.COLUMNNAME_NOT_EDIT_SHOW,
'parent_menu_id': int(gen_table.parent_menu_id) if gen_table.parent_menu_id else None,
}
return context
@classmethod
def get_template_list(cls):
"""
获取模板列表
参数:
-
返回:
- List[str]: 模板路径列表
"""
templates = [
'python/controller.py.j2',
'python/service.py.j2',
'python/crud.py.j2',
'python/schema.py.j2',
'python/model.py.j2',
'python/__init__.py.j2',
'ts/api.ts.j2',
'vue/index.vue.j2',
]
return templates
@classmethod
def get_file_name(cls, template: str, gen_table: GenTableOutSchema):
"""
根据模板生成文件名
参数:
- template (str): 模板路径字符串
- gen_table (GenTableOutSchema): 生成表的配置信息
返回:
- str: 模板生成的文件名
异常:
- ValueError: 当无法生成有效文件名时抛出
"""
module_name = gen_table.module_name or ''
business_name = gen_table.business_name or ''
# 验证必要的参数
if not module_name or not business_name:
raise ValueError(f"无法为模板 {template} 生成文件名:模块名或业务名未设置")
# 映射表方式简化
template_mapping = {
'controller.py.j2': f'{cls.BACKEND_PROJECT_PATH}/app/plugin/{module_name}/{business_name}/controller.py',
'service.py.j2': f'{cls.BACKEND_PROJECT_PATH}/app/plugin/{module_name}/{business_name}/service.py',
'crud.py.j2': f'{cls.BACKEND_PROJECT_PATH}/app/plugin/{module_name}/{business_name}/crud.py',
'schema.py.j2': f'{cls.BACKEND_PROJECT_PATH}/app/plugin/{module_name}/{business_name}/schema.py',
'model.py.j2': f'{cls.BACKEND_PROJECT_PATH}/app/plugin/{module_name}/{business_name}/model.py',
'__init__.py.j2': f'{cls.BACKEND_PROJECT_PATH}/app/plugin/{module_name}/{business_name}/__init__.py',
'api.ts.j2': f'{cls.FRONTEND_PROJECT_PATH}/src/api/{module_name}/{business_name}.ts',
'index.vue.j2': f'{cls.FRONTEND_PROJECT_PATH}/src/views/{module_name}/{business_name}/index.vue'
}
# 查找匹配的模板路径
for key, path in template_mapping.items():
if key in template:
return path
# 遍历完所有映射都没找到匹配项,才抛出异常
raise ValueError(f"未找到模板 '{template}' 的路径映射")
@classmethod
def get_package_prefix(cls, package_name: str) -> str:
"""
获取包前缀
参数:
- package_name (str): 包名
返回:
- str: 包前缀
"""
# 修复:当包名中不存在'.'时,直接返回原包名
return package_name[: package_name.rfind('.')] if '.' in package_name else package_name
@classmethod
def get_schema_import_list(cls, gen_table: GenTableOutSchema):
"""
获取schema模板导入包列表
:param gen_table: 生成表的配置信息
:return: 导入包列表
"""
columns = gen_table.columns or []
import_list = set()
has_datetime_type = False
for column in columns:
# 处理嵌套的datetime类型,如datetime.date、datetime.time、datetime.datetime
if column.python_type.startswith('datetime.') or column.python_type in GenConstant.TYPE_DATE:
has_datetime_type = True
elif column.python_type == GenConstant.TYPE_DECIMAL:
import_list.add('from decimal import Decimal')
if gen_table.sub:
if gen_table.sub_table and gen_table.sub_table.columns:
sub_columns = gen_table.sub_table.columns or []
for sub_column in sub_columns:
# 处理嵌套的datetime类型,如datetime.date、datetime.time、datetime.datetime
if sub_column.python_type.startswith('datetime.') or sub_column.python_type in GenConstant.TYPE_DATE:
has_datetime_type = True
elif sub_column.python_type == GenConstant.TYPE_DECIMAL:
import_list.add('from decimal import Decimal')
if has_datetime_type:
import_list.add('import datetime')
return import_list
@classmethod
def get_model_import_list(cls, gen_table: GenTableOutSchema):
"""
获取do模板导入包列表
:param gen_table: 生成表的配置信息
:return: 导入包列表
"""
columns = gen_table.columns or []
import_list = set()
for column in columns:
if column.column_type:
data_type = cls.get_db_type(column.column_type)
if data_type in GenConstant.COLUMNTYPE_GEOMETRY:
import_list.add('from geoalchemy2 import Geometry')
import_list.add(
f'from sqlalchemy import {StringUtil.get_mapping_value_by_key_ignore_case(GenConstant.DB_TO_SQLALCHEMY, data_type)}'
)
# 处理datetime类型的导入
if column.python_type and '.' in column.python_type:
datetime_type = column.python_type.split('.')[0]
if datetime_type == 'datetime':
import_list.add('import datetime')
if gen_table.sub:
import_list.add('from sqlalchemy import ForeignKey')
if gen_table.sub_table and gen_table.sub_table.columns:
sub_columns = gen_table.sub_table.columns or []
for sub_column in sub_columns:
if sub_column.column_type:
data_type = cls.get_db_type(sub_column.column_type)
import_list.add(
f'from sqlalchemy import {StringUtil.get_mapping_value_by_key_ignore_case(GenConstant.DB_TO_SQLALCHEMY, data_type)}'
)
# 处理datetime类型的导入
if sub_column.python_type and '.' in sub_column.python_type:
datetime_type = sub_column.python_type.split('.')[0]
if datetime_type == 'datetime':
import_list.add('import datetime')
return cls.merge_same_imports(list(import_list), 'from sqlalchemy import')
@classmethod
def get_db_type(cls, column_type: str) -> str:
"""
获取数据库字段类型
参数:
- column_type (str): 字段类型字符串
返回:
- str: 数据库类型去除长度等修饰
"""
if '(' in column_type:
return column_type.split('(')[0]
return column_type
@classmethod
def merge_same_imports(cls, imports: list[str], import_start: str) -> list[str]:
"""
合并相同的导入语句
参数:
- imports (list[str]): 导入语句列表
- import_start (str): 导入语句的起始字符串
返回:
- list[str]: 合并后的导入语句列表
"""
merged_imports = []
_imports = []
for import_stmt in imports:
if import_stmt.startswith(import_start):
imported_items = import_stmt.split('import')[1].strip()
_imports.extend(imported_items.split(', '))
else:
merged_imports.append(import_stmt)
if _imports:
merged_datetime_import = f'{import_start} {", ".join(_imports)}'
merged_imports.append(merged_datetime_import)
return merged_imports
@classmethod
def get_dicts(cls, gen_table: GenTableOutSchema):
"""
获取字典列表
参数:
- gen_table (GenTableOutSchema): 生成表的配置信息
返回:
- str: 以逗号分隔的字典类型字符串
"""
columns = gen_table.columns or []
dicts = set()
cls.add_dicts(dicts, columns)
# 处理sub_table为None的情况
if gen_table.sub_table is not None:
# 处理sub_table.columns为None的情况
sub_columns = gen_table.sub_table.columns or []
cls.add_dicts(dicts, sub_columns)
return ', '.join(dicts)
@classmethod
def add_dicts(cls, dicts: set[str], columns: list[GenTableColumnOutSchema]):
"""
添加字典类型到集合
参数:
- dicts (set[str]): 字典类型集合
- columns (list[GenTableColumnOutSchema]): 字段列表
返回:
- set[str]: 更新后的字典类型集合
"""
for column in columns:
super_column = column.super_column if column.super_column is not None else '0'
dict_type = column.dict_type or ''
html_type = column.html_type or ''
if (
not super_column
and StringUtil.is_not_empty(dict_type)
and StringUtil.equals_any_ignore_case(
html_type, [GenConstant.HTML_SELECT, GenConstant.HTML_RADIO, GenConstant.HTML_CHECKBOX]
)
):
dicts.add(f"'{dict_type}'")
@classmethod
def get_permission_prefix(cls, module_name: str | None, business_name: str | None) -> str:
"""
获取权限前缀
参数:
- module_name (str | None): 模块名
- business_name (str | None): 业务名
返回:
- str: 权限前缀字符串
"""
return f'{module_name}:{business_name}'
@classmethod
def get_sqlalchemy_type(cls, column):
"""
获取 SQLAlchemy 类型
参数:
- column (Any): 列对象或列类型字符串
返回:
- str: SQLAlchemy 类型字符串
"""
# 获取column_type和column_length
column_type = column
column_length = None
# 检查是否是对象
if hasattr(column, 'column_type'):
column_type = column.column_type or ''
column_length = column.column_length or None
# 首先尝试匹配完整类型(包括括号)
sqlalchemy_type = StringUtil.get_mapping_value_by_key_ignore_case(
GenConstant.DB_TO_SQLALCHEMY, column_type
)
if sqlalchemy_type is None and '(' in column_type:
# 如果没有匹配到,再尝试剥离括号
column_type_list = column_type.split('(')
col_type = column_type_list[0]
# 将 'character' 映射为 'char' 以匹配常量定义
if col_type.lower() == 'character':
col_type = 'char'
sqlalchemy_type = StringUtil.get_mapping_value_by_key_ignore_case(
GenConstant.DB_TO_SQLALCHEMY, col_type
)
# 如果是字符串类型且包含括号参数,保持原参数
if sqlalchemy_type in ["String", "CHAR"]:
sqlalchemy_type += '(' + column_type_list[1]
elif sqlalchemy_type is None:
# 处理没有括号的类型
col_type = column_type
# 将 'character' 映射为 'char' 以匹配常量定义
if col_type.lower() == 'character':
col_type = 'char'
sqlalchemy_type = StringUtil.get_mapping_value_by_key_ignore_case(
GenConstant.DB_TO_SQLALCHEMY, col_type
)
# 如果是字符串类型且没有指定长度,使用column_length或默认255
if sqlalchemy_type in ["String", "CHAR"]:
length = column_length if column_length and column_length.isdigit() else '255'
sqlalchemy_type += f'({length})'
else:
# 对于已经匹配到的类型,如果是字符串类型且column有长度信息,添加长度
if sqlalchemy_type in ["String", "CHAR"] and not '(' in sqlalchemy_type:
# 检查column_length是否有效
length = column_length if column_length and column_length.isdigit() else '255'
sqlalchemy_type += f'({length})'
# 如果没有找到匹配的类型,使用String(column_length)或String(255)作为默认类型
if not sqlalchemy_type:
length = column_length if column_length and column_length.isdigit() else '255'
sqlalchemy_type = f'String({length})'
return sqlalchemy_type