refactor(gencode): 重构代码生成模块的schema和crud层

- 合并GenTableCreateSchema和GenTableUpdateSchema为GenTableBaseSchema
- 移除不必要的OutSchema类,简化模型结构
- 优化数据库查询方法,使用更安全的SQL构建方式
- 统一列查询接口,增强类型安全性
This commit is contained in:
zhangtao
2025-10-03 05:06:29 +08:00
parent 119964e6f9
commit 609324f6c7
2 changed files with 306 additions and 305 deletions
@@ -1,26 +1,28 @@
# -*- coding:utf-8 -*-
from datetime import datetime, time
import json
from sqlalchemy.engine.row import Row
from sqlalchemy import delete, func, select, text, update
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import and_, delete, select, text, update
from sqlalchemy.orm import selectinload
from typing import List, Optional, Sequence, Any, Dict
from typing import List, Optional, Sequence, Dict
from app.api.v1.module_system.params.schema import ParamsCreateSchema
from app.core.logger import logger
from .model import GenTableModel, GenTableColumnModel
from app.config.setting import settings
from app.common.request import PaginationService
from .schema import GenTableCreateSchema, GenTableUpdateSchema, GenTableOutSchema, GenTableDeleteSchema, GenTableColumnCreateSchema, GenTableColumnUpdateSchema, GenTableColumnOutSchema, GenTableColumnDeleteSchema, GenDBTableSchema
from .param import GenTableQueryParam
from .schema import (
GenTableSchema,
GenTableDeleteSchema,
GenTableColumnSchema,
GenTableColumnDeleteSchema,
GenDBTableSchema,
)
from .param import GenTableQueryParam, GenTableColumnQueryParam
from app.core.base_crud import CRUDBase
from app.api.v1.module_system.auth.schema import AuthSchema
class GenTableCRUD(CRUDBase[GenTableModel, GenTableCreateSchema, GenTableUpdateSchema]):
class GenTableCRUD(CRUDBase[GenTableModel, GenTableSchema, GenTableSchema]):
"""代码生成业务表模块数据库操作层"""
def __init__(self, auth: AuthSchema) -> None:
@@ -69,21 +71,6 @@ class GenTableCRUD(CRUDBase[GenTableModel, GenTableCreateSchema, GenTableUpdateS
return gen_table
async def get_gen_table_all(self) -> Sequence[GenTableModel]:
"""
获取所有业务表信息
:param db: orm对象
:return: 所有业务表信息
"""
gen_table_all = (
await self.db.execute(
select(GenTableModel)
.options(selectinload(GenTableModel.columns)))
).scalars().all()
return gen_table_all
async def get_gen_table_list(self, search: Optional[GenTableQueryParam] = None):
"""
根据查询参数获取代码生成业务表列表信息
@@ -96,9 +83,7 @@ class GenTableCRUD(CRUDBase[GenTableModel, GenTableCreateSchema, GenTableUpdateS
query = (
select(GenTableModel)
.options(selectinload(GenTableModel.columns))
.where(
*conditions
)
.where(*conditions)
.order_by(GenTableModel.created_at.desc())
.distinct()
)
@@ -109,6 +94,199 @@ class GenTableCRUD(CRUDBase[GenTableModel, GenTableCreateSchema, GenTableUpdateS
return gen_table_all
async def add_gen_table(self, add_model: GenTableSchema) -> GenTableModel:
"""
增加
"""
gen_table = GenTableModel(
**add_model.model_dump(exclude_unset=True, exclude={"sub", "tree", "crud"})
)
self.db.add(gen_table)
await self.db.flush()
return gen_table
async def edit_gen_table(self, table_id: int, edit_model: GenTableSchema):
"""
修改
"""
edit_dict_data = edit_model.model_dump(exclude_unset=True)
await self.db.execute(
update(GenTableModel)
.where(GenTableModel.id == table_id)
.values(**edit_dict_data)
)
await self.db.flush()
await self.db.commit()
return edit_model
async def delete_gen_table(self, delete_model: GenTableDeleteSchema) -> None:
"""
删除
"""
await self.db.execute(
delete(GenTableModel).where(GenTableModel.id.in_(delete_model.table_ids))
)
await self.db.flush()
async def get_db_table_list(self, search: Optional[GenTableQueryParam] = None) -> list[Dict]:
"""
根据查询参数获取数据库列表信息
:param search: 查询参数对象
:return: 数据库列表信息对象
"""
# 使用更健壮的方式检测数据库方言
if settings.DATABASE_TYPE == "postgresql":
query_sql = (
select(
text("table_catalog as database_name"),
text("table_name as table_name"),
text("table_type as table_type"),
text("table_comment as table_comment"),
)
.select_from(text("information_schema.tables"))
.where(
and_(
text("table_catalog = (select current_database())"),
text("is_insertable_into = 'YES'"),
text("table_schema = 'public'"),
)
)
)
elif settings.DATABASE_TYPE == "mysql":
query_sql = (
select(
text("table_schema as database_name"),
text("table_name as table_name"),
text("table_type as table_type"),
text("table_comment as table_comment"),
)
.select_from(text("information_schema.tables"))
.where(
and_(
text("table_schema = (select database())"),
)
)
)
else:
query_sql = (
select(
text(f"{settings.DATABASE_NAME} as database_name"),
text("name as table_name"),
text("type as table_type"),
text("tbl_name as table_comment"),
)
.select_from(text("sqlite_master"))
.where(
and_(
text("type = 'table'"),
)
)
)
# 动态条件构造
if search and search.table_name:
query_sql = query_sql.where(
text("lower(table_name) like lower(:table_name)")
)
if search and search.table_comment:
query_sql = query_sql.where(
text("lower(table_comment) like lower(:table_comment)")
)
# 执行查询
all_data =(await self.db.execute(query_sql)).fetchall()
# 将Row对象转换为字典列表,解决JSON序列化问题
dict_data = []
for row in all_data:
# 检查row是否为Row对象
if isinstance(row, Row):
# 使用._mapping获取字典
dict_row = GenDBTableSchema(**dict(row._mapping)).model_dump()
dict_data.append(dict_row)
else:
dict_row = GenDBTableSchema(**dict(row)).model_dump()
dict_data.append(dict_row)
return dict_data
async def get_db_table_list_by_names(self, table_names: List[str]) -> list[Dict]:
"""
根据业务表名称组获取数据库列表信息
:param table_names: 业务表名称组
:return: 数据库列表信息对象
"""
# 使用更健壮的方式检测数据库方言
if settings.DATABASE_TYPE == "postgresql":
query_sql = (
select(
text("table_catalog as database_name"),
text("table_name as table_name"),
text("table_type as table_type"),
text("table_comment as table_comment"),
)
.select_from(text("information_schema.tables"))
.where(
and_(
text("table_catalog = (select current_database())"),
text("is_insertable_into = 'YES'"),
text("table_schema = 'public'"),
)
)
)
elif settings.DATABASE_TYPE == "mysql":
query_sql = (
select(
text("table_schema as database_name"),
text("table_name as table_name"),
text("table_type as table_type"),
text("table_comment as table_comment"),
)
.select_from(text("information_schema.tables"))
.where(
and_(
text("table_schema = (select database())"),
)
)
)
else:
query_sql = (
select(
text(f"{settings.DATABASE_NAME} as database_name"),
text("name as table_name"),
text("type as table_type"),
text("tbl_name as table_comment"),
)
.select_from(text("sqlite_master"))
.where(
and_(
text("type = 'table'"),
)
)
)
query_sql = query_sql.where(
text(f"table_name in :{table_names}")
)
gen_db_table_list = (await self.db.execute(query_sql)).fetchall()
# 将Row对象转换为字典列表,解决JSON序列化问题
dict_data = []
for row in gen_db_table_list:
# 检查row是否为Row对象
if isinstance(row, Row):
# 使用._mapping获取字典
dict_row = GenDBTableSchema(**dict(row._mapping)).model_dump()
dict_data.append(dict_row)
else:
dict_row = GenDBTableSchema(**dict(row)).model_dump()
dict_data.append(dict_row)
return dict_data
async def create_table_by_sql(self, sql: str) -> bool:
"""
根据sql语句创建表结构
@@ -117,7 +295,6 @@ class GenTableCRUD(CRUDBase[GenTableModel, GenTableCreateSchema, GenTableUpdateS
:param sql_statements: sql语句的ast列表
:return:
"""
await self.db.execute(text(sql))
try:
await self.db.execute(text(sql))
# 提交事务
@@ -130,163 +307,8 @@ class GenTableCRUD(CRUDBase[GenTableModel, GenTableCreateSchema, GenTableUpdateS
logger.error(f"创建表时发生错误: {e}")
return False
async def add_gen_table(self, add_model: GenTableCreateSchema) -> GenTableModel:
"""
增加
"""
gen_table = GenTableModel(**add_model.model_dump(exclude_unset=True, exclude={'sub', 'tree', 'crud'}))
self.db.add(gen_table)
await self.db.flush()
return gen_table
async def delete_gen_table(self, delete_model: GenTableDeleteSchema) -> None:
"""
删除
"""
await self.db.execute(delete(GenTableModel).where(GenTableModel.id.in_(delete_model.table_ids)))
await self.db.flush()
async def edit_gen_table(self, table_id: int, edit_model: GenTableUpdateSchema, auto_commit: bool = True):
"""
修改
"""
edit_dict_data = edit_model.model_dump(exclude_unset=True)
await self.db.execute(update(GenTableModel).where(GenTableModel.id == table_id).values(**edit_dict_data))
await self.db.flush()
if auto_commit:
await self.db.commit()
return edit_model
async def get_gen_db_table_list(self, table_name: Optional[str] = None) -> list[Any]:
"""
根据查询参数获取数据库列表信息
:param db: orm对象
:param search: 查询参数对象
:param order_by: 排序字段
:return: 数据库列表信息对象
"""
# 使用更健壮的方式检测数据库方言
if settings.DATABASE_TYPE == 'postgresql':
query_sql = """
SELECT
table_catalog as database_name,
table_name as table_name,
table_type as table_type,
table_schema as table_comment
from
information_schema.tables
where
table_catalog = (select current_database())
and is_insertable_into = 'YES'
and table_schema = 'public'
"""
elif settings.DATABASE_TYPE == 'mysql':
query_sql = """
SELECT
table_schema as database_name,
table_name as table_name,
table_type as table_type,
table_comment as table_comment
from
information_schema.tables
where
table_schema = (select database())
"""
else:
query_sql = f"""
SELECT
'{settings.DATABASE_NAME}' as database_name,
name as table_name,
type as table_type,
tbl_name as table_comment
from
sqlite_master
where
type = 'table'
"""
# 直接执行文本SQL查询,避免SQLAlchemy自动添加额外的SELECT )
query = text(query_sql).bindparams()
# 执行查询
result = await self.db.execute(query)
all_data = result.fetchall()
# 将Row对象转换为字典列表,解决JSON序列化问题
dict_data = []
for row in all_data:
# 检查row是否为Row对象
if isinstance(row, Row):
# 使用._mapping获取字典
dict_row = GenDBTableSchema(**dict(row._mapping)).model_dump()
if table_name:
dict_row['table_name'] = table_name
dict_data.append(dict_row)
else:
dict_row = GenDBTableSchema(**dict(row)).model_dump()
dict_data.append(dict_row)
return dict_data
async def get_gen_db_table_list_by_names(self, table_names: List[str]):
"""
根据业务表名称组获取数据库列表信息
:param db: orm对象
:param table_names: 业务表名称组
:return: 数据库列表信息对象
"""
# 使用更健壮的方式检测数据库方言
if settings.DATABASE_TYPE == 'postgresql':
query_sql = """
SELECT
table_catalog as database_name,
table_name as table_name,
table_type as table_type,
table_schema as table_comment
from
information_schema.tables
where
table_catalog = (select current_database())
and is_insertable_into = 'YES'
and table_schema = 'public'
and table_name in :table_names
"""
elif settings.DATABASE_TYPE == 'mysql':
query_sql = """
SELECT
table_schema as database_name,
table_name as table_name,
table_type as table_type,
table_comment as table_comment
from
information_schema.tables
where
table_schema = (select database())
and table_name in :table_names
"""
else:
query_sql = f"""
SELECT
'{settings.DATABASE_NAME}' as database_name,
name as table_name,
type as table_type,
tbl_name as table_comment
from
sqlite_master
where
type = 'table'
and table_name in :table_names
"""
query = text(query_sql).bindparams(table_names=tuple(table_names))
gen_db_table_list = (await self.db.execute(query)).fetchall()
return gen_db_table_list
class GenTableColumnCRUD(CRUDBase[GenTableColumnModel, GenTableColumnCreateSchema, GenTableColumnUpdateSchema]):
class GenTableColumnCRUD(CRUDBase[GenTableColumnModel, GenTableColumnSchema, GenTableColumnSchema]):
"""代码生成业务表字段模块数据库操作层"""
def __init__(self, auth: AuthSchema) -> None:
@@ -297,15 +319,23 @@ class GenTableColumnCRUD(CRUDBase[GenTableColumnModel, GenTableColumnCreateSchem
"""详情"""
return await self.get(id=column_id)
async def get_list_crud(self, search: Optional[Dict] = None, order_by: Optional[List[Dict[str, str]]] = None) -> Sequence[GenTableColumnModel]:
async def list_crud(
self,
search: Optional[Dict] = None,
order_by: Optional[List[Dict[str, str]]] = None,
) -> Sequence[GenTableColumnModel]:
"""列表查询"""
return await self.list(search=search, order_by=order_by)
async def create_crud(self, data: GenTableColumnCreateSchema) -> Optional[GenTableColumnModel]:
async def create_crud(
self, data: GenTableColumnSchema
) -> Optional[GenTableColumnModel]:
"""创建"""
return await self.create(data=data)
async def update_crud(self, id: int, data: GenTableColumnUpdateSchema) -> Optional[GenTableColumnModel]:
async def update_crud(
self, id: int, data: GenTableColumnSchema
) -> Optional[GenTableColumnModel]:
"""更新"""
return await self.update(id=id, data=data)
@@ -313,93 +343,83 @@ class GenTableColumnCRUD(CRUDBase[GenTableColumnModel, GenTableColumnCreateSchem
"""批量删除"""
return await self.delete(ids=data.column_ids)
async def get_gen_table_column_list_by_table_id_crud(self, table_id: int) -> Sequence[GenTableColumnModel]:
"""根据业务表id获取需要生成的业务表字段列表信息"""
return await self.list(search={"table_id": table_id})
async def get_gen_table_column_list_by_table_id(self, table_id: int) -> Sequence[GenTableColumnModel]:
"""
根据业务表id获取需要生成的业务表字段列表信息
:param table_id: 业务表id
:return: 需要生成的业务表字段列表信息对象
"""
gen_table_column_list = (
(
await self.db.execute(
select(GenTableColumnModel)
.where(GenTableColumnModel.table_id == table_id)
.order_by(GenTableColumnModel.sort)
)
)
.scalars()
.all()
)
return gen_table_column_list
async def get_gen_db_table_columns_by_name(self, db: AsyncSession, table_name: str) -> Sequence[Row[Any]]:
async def get_gen_db_table_columns_by_name(self, table_name: str) -> List[GenTableColumnSchema]:
"""
根据业务表名称获取业务表字段列表信息
:param db: orm对象
:param table_name: 业务表名称
:return: 业务表字段列表信息对象
"""
# 检查表名是否为空
if not table_name:
raise ValueError("数据表名称不能为空")
# 兼容SQLite和MySQL/PostgreSQL
if str(db.bind.dialect) == 'sqlite':
if settings.DATABASE_TYPE == "postgresql":
query_sql = """
pragma table_info(:table_name)
SELECT
column_name,
(CASE WHEN (is_nullable = 'no' AND column_key != 'PRI') THEN '1' ELSE '0' END) AS is_required,
(CASE WHEN column_key = 'PRI' THEN '1' ELSE '0' END) AS is_pk,
ordinal_position AS sort,
column_comment,
(CASE WHEN extra = 'auto_increment' THEN '1' ELSE '0' END) AS is_increment,
column_type
FROM
information_schema.tables
WHERE
table_catalog = (select current_database())
AND is_insertable_into = 'YES'
AND table_schema = 'public'
AND table_name = :table_name
"""
elif settings.DATABASE_TYPE == "mysql":
query_sql = """
SELECT
column_name,
(CASE WHEN (is_nullable = 'no' AND column_key != 'PRI') THEN '1' ELSE '0' END) AS is_required,
(CASE WHEN column_key = 'PRI' THEN '1' ELSE '0' END) AS is_pk,
ordinal_position AS sort,
column_comment,
(CASE WHEN extra = 'auto_increment' THEN '1' ELSE '0' END) AS is_increment,
column_type
FROM
information_schema.tables
WHERE
table_schema = (SELECT DATABASE())
AND table_name = :table_name
"""
query = text(query_sql).bindparams(table_name=table_name)
gen_db_table_columns_raw = (await db.execute(query)).fetchall()
# 转换SQLite的pragma结果为与information_schema.columns兼容的格式
gen_db_table_columns = []
for col in gen_db_table_columns_raw:
# col格式: (cid, name, type, notnull, dflt_value, pk)
is_required = '1' if col[3] == 1 and col[5] == 0 else '0'
is_pk = '1' if col[5] == 1 else '0'
is_increment = '0' # SQLite没有auto_increment标记,需要额外判断
# 构建兼容的结果行
gen_db_table_columns.append({
'column_name': col[1],
'is_required': is_required,
'is_pk': is_pk,
'sort': col[0], # 使用cid作为排序
'column_comment': '', # SQLite不存储列注释
'is_increment': is_increment,
'column_type': col[2]
})
else:
query_sql = """
select
column_name as column_name,
case
when is_nullable = 'no' and column_key != 'PRI' then '1'
else '0'
end as is_required,
case
when column_key = 'PRI' then '1'
else '0'
end as is_pk,
ordinal_position as sort,
column_comment as column_comment,
case
when extra = 'auto_increment' then '1'
else '0'
end as is_increment,
column_type as column_type
from
information_schema.columns
where
table_schema = (select database())
and table_name = :table_name
order by
ordinal_position
query_sql = f"""
SELECT
column_name,
(CASE WHEN (is_nullable = 'no' AND column_key != 'PRI') THEN '1' ELSE '0' END) AS is_required,
(CASE WHEN column_key = 'PRI' THEN '1' ELSE '0' END) AS is_pk,
ordinal_position AS sort,
column_comment,
(CASE WHEN extra = 'auto_increment' THEN '1' ELSE '0' END) AS is_increment,
column_type
FROM
sqlite_master
WHERE
type = 'table'
AND name = :table_name
"""
query = text(query_sql).bindparams(table_name=table_name)
gen_db_table_columns = (await db.execute(query)).fetchall()
return gen_db_table_columns
query = text(query_sql).bindparams(table_name=table_name)
gen_db_table_columns_raw = (
await self.db.execute(query)
).fetchall()
return [
GenTableColumnSchema(
column_name=row[0],
is_required=row[1],
is_pk=row[2],
sort=row[3],
column_comment=row[4],
is_increment=row[5],
column_type=row[6],
)
for row in gen_db_table_columns_raw
]
@@ -1,6 +1,6 @@
# -*- coding:utf-8 -*-
from typing import List, Literal, Optional
from typing import Any, List, Literal, Optional
from pydantic import BaseModel, ConfigDict, Field, model_validator
from pydantic.alias_generators import to_camel
@@ -28,7 +28,7 @@ class GenDBTableSchema(BaseModel):
table_comment: Optional[str] = Field(default=None, description='表描述')
class GenTableCreateSchema(BaseModel):
class GenTableBaseSchema(BaseModel):
"""
代码生成业务表创建模型
"""
@@ -51,14 +51,16 @@ class GenTableCreateSchema(BaseModel):
options: Optional[str] = Field(default=None, description='其它生成选项')
description: Optional[str] = Field(default=None, description='功能描述')
params: Optional[Any] = Field(default=None, description='前端传递过来的表附加信息,转换成json字符串后放到options')
class GenTableUpdateSchema(GenTableCreateSchema):
class GenTableSchema(GenTableBaseSchema):
"""
代码生成业务表更新模型
"""
pk_column: Optional['GenTableColumnUpdateSchema'] = Field(default=None, description='主键信息')
sub_table: Optional['GenTableUpdateSchema'] = Field(default=None, description='子表信息')
columns: Optional[List['GenTableColumnUpdateSchema']] = Field(default=None, description='表列信息')
pk_column: Optional['GenTableColumnSchema'] = Field(default=None, description='主键信息')
sub_table: Optional['GenTableSchema'] = Field(default=None, description='子表信息')
columns: Optional[List['GenTableColumnSchema']] = Field(default=None, description='表列信息')
tree_code: Optional[str] = Field(default=None, description='树编码字段tree_code')
tree_parent_code: Optional[str] = Field(default=None, description='树父编码字段')
tree_name: Optional[str] = Field(default=None, description='树名称字段ree_name')
@@ -69,20 +71,13 @@ class GenTableUpdateSchema(GenTableCreateSchema):
crud: Optional[bool] = Field(default=None, description='是否为单表')
@model_validator(mode='after')
def check_some_is(self) -> 'GenTableUpdateSchema':
def check_some_is(self) -> 'GenTableSchema':
self.sub = True if self.tpl_category and self.tpl_category == GenConstant.TPL_SUB else False
self.tree = True if self.tpl_category and self.tpl_category == GenConstant.TPL_TREE else False
self.crud = True if self.tpl_category and self.tpl_category == GenConstant.TPL_CRUD else False
return self
class GenTableOutSchema(GenTableUpdateSchema, BaseSchema):
"""
代码生成业务表响应模型
"""
model_config = ConfigDict(from_attributes=True)
class GenTableDeleteSchema(BaseModel):
"""
删除代码生成业务表模型
@@ -92,7 +87,7 @@ class GenTableDeleteSchema(BaseModel):
table_ids: str = Field(..., description='需要删除的代码生成业务表ID')
class GenTableColumnCreateSchema(BaseModel):
class GenTableColumnSchema(BaseModel):
"""
代码生成业务表字段创建模型
"""
@@ -119,20 +114,6 @@ class GenTableColumnCreateSchema(BaseModel):
description: Optional[str] = Field(default=None, description='功能描述')
class GenTableColumnUpdateSchema(GenTableColumnCreateSchema):
"""
代码生成业务表字段更新模型
"""
...
class GenTableColumnOutSchema(GenTableColumnUpdateSchema, BaseSchema):
"""
代码生成业务表字段响应模型
"""
model_config = ConfigDict(from_attributes=True)
class GenTableColumnDeleteSchema(BaseModel):
"""
删除代码生成业务表字段模型