From 609324f6c70bec2a99467bdc65a515c928710b3a Mon Sep 17 00:00:00 2001 From: zhangtao <9480807882@qq.com> Date: Fri, 3 Oct 2025 05:06:29 +0800 Subject: [PATCH] =?UTF-8?q?refactor(gencode):=20=E9=87=8D=E6=9E=84?= =?UTF-8?q?=E4=BB=A3=E7=A0=81=E7=94=9F=E6=88=90=E6=A8=A1=E5=9D=97=E7=9A=84?= =?UTF-8?q?schema=E5=92=8Ccrud=E5=B1=82?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 合并GenTableCreateSchema和GenTableUpdateSchema为GenTableBaseSchema - 移除不必要的OutSchema类，简化模型结构 - 优化数据库查询方法，使用更安全的SQL构建方式 - 统一列查询接口，增强类型安全性 --- .../api/v1/module_generator/gencode/crud.py | 572 +++++++++--------- .../api/v1/module_generator/gencode/schema.py | 39 +- 2 files changed, 306 insertions(+), 305 deletions(-) diff --git a/backend/app/api/v1/module_generator/gencode/crud.py b/backend/app/api/v1/module_generator/gencode/crud.py index f170f1f2..3b2b4551 100644 --- a/backend/app/api/v1/module_generator/gencode/crud.py +++ b/backend/app/api/v1/module_generator/gencode/crud.py @@ -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,19 +83,210 @@ class GenTableCRUD(CRUDBase[GenTableModel, GenTableCreateSchema, GenTableUpdateS query = ( select(GenTableModel) .options(selectinload(GenTableModel.columns)) - .where( - *conditions - ) + .where(*conditions) .order_by(GenTableModel.created_at.desc()) .distinct() ) - + # 获取所有数据 result = await self.db.execute(query) gen_table_all = list(result.scalars().all()) - + 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: @@ -296,110 +318,108 @@ class GenTableColumnCRUD(CRUDBase[GenTableColumnModel, GenTableColumnCreateSchem async def get_by_id_crud(self, column_id: int) -> Optional[GenTableColumnModel]: """详情""" 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) - + async def delete_crud(self, data: GenTableColumnDeleteSchema) -> None: """批量删除""" 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: 业务表字段列表信息对象 """ - # 兼容SQLite和MySQL/PostgreSQL - if str(db.bind.dialect) == 'sqlite': - query_sql = """ - pragma table_info(: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 = text(query_sql).bindparams(table_name=table_name) - gen_db_table_columns = (await db.execute(query)).fetchall() + # 检查表名是否为空 + if not table_name: + raise ValueError("数据表名称不能为空") - return gen_db_table_columns \ No newline at end of file + # 兼容SQLite和MySQL/PostgreSQL + if settings.DATABASE_TYPE == "postgresql": + 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_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 + """ + else: + 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_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 + ] diff --git a/backend/app/api/v1/module_generator/gencode/schema.py b/backend/app/api/v1/module_generator/gencode/schema.py index bc9656c7..7922eb5c 100644 --- a/backend/app/api/v1/module_generator/gencode/schema.py +++ b/backend/app/api/v1/module_generator/gencode/schema.py @@ -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): """ 删除代码生成业务表字段模型