"""数据字典服务 — 通过 DBConnector 连接数据源获取元数据。""" import re from typing import List, Dict, Any, Optional from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from app.models.instance import Instance from app.services.db_connector import DBConnector from app.core.crypto import decrypt_password async def _get_instance(session: AsyncSession, instance_id: int) -> Instance: """根据 ID 查询实例,解密密码后返回。""" result = await session.execute(select(Instance).where(Instance.id == instance_id)) instance = result.scalar_one_or_none() if not instance: return None instance.password = decrypt_password(instance.password) return instance async def list_databases(session: AsyncSession, instance_id: int) -> Optional[List[str]]: """获取实例的数据库列表。""" instance = await _get_instance(session, instance_id) if not instance: return None if instance.db_type == "mysql": qr = await DBConnector.execute_query( instance, None, "SHOW DATABASES", user_id=0 ) system_dbs = {"information_schema", "mysql", "performance_schema", "sys"} return [r.get("Database", r.get("database", "")) for r in qr.rows if r.get("Database", r.get("database", "")) not in system_dbs] elif instance.db_type in ("postgres", "postgresql"): qr = await DBConnector.execute_query( instance, "postgres", "SELECT datname FROM pg_database WHERE datistemplate = false", user_id=0, ) return [r.get("datname", "") for r in qr.rows] elif instance.db_type == "sqlite": return [instance.database or "main"] else: return [] async def list_tables( session: AsyncSession, instance_id: int, database: str ) -> List[str]: """获取指定数据库的表名列表。""" instance = await _get_instance(session, instance_id) if not instance: return [] if instance.db_type == "mysql": sql = ( "SELECT TABLE_NAME FROM information_schema.TABLES " "WHERE TABLE_SCHEMA = '" + database + "' AND TABLE_TYPE = 'BASE TABLE'" ) qr = await DBConnector.execute_query(instance, database, sql, user_id=0) return [r.get("TABLE_NAME", "") for r in qr.rows] elif instance.db_type in ("postgres", "postgresql"): sql = "SELECT tablename FROM pg_tables WHERE schemaname = 'public'" qr = await DBConnector.execute_query(instance, database, sql, user_id=0) return [r.get("tablename", "") for r in qr.rows] elif instance.db_type == "sqlite": sql = "SELECT name FROM sqlite_master WHERE type='table'" qr = await DBConnector.execute_query(instance, None, sql, user_id=0) return [r.get("name", "") for r in qr.rows] else: return [] async def get_table_detail( session: AsyncSession, instance_id: int, database: str, table_name: str ) -> Dict[str, Any]: """获取表详情: columns / indexes / create_ddl / row_count。""" instance = await _get_instance(session, instance_id) if not instance: return {} columns = [] indexes = [] create_ddl = "" row_count = 0 if instance.db_type == "mysql": # Columns col_sql = ( "SELECT COLUMN_NAME as Field, COLUMN_TYPE as Type, " "IS_NULLABLE as `Null`, COLUMN_KEY as `Key`, " "COLUMN_DEFAULT as `Default`, EXTRA as Extra, " "COLUMN_COMMENT as Comment " "FROM information_schema.COLUMNS " "WHERE TABLE_SCHEMA = '" + database + "' " "AND TABLE_NAME = '" + table_name + "' " "ORDER BY ORDINAL_POSITION" ) col_qr = await DBConnector.execute_query(instance, database, col_sql, user_id=0) columns = col_qr.rows # Indexes idx_sql = "SHOW INDEX FROM `" + table_name + "`" idx_qr = await DBConnector.execute_query(instance, database, idx_sql, user_id=0) # Group by index name idx_map = {} for r in idx_qr.rows: idx_name = r.get("Key_name", "") if idx_name not in idx_map: idx_map[idx_name] = { "name": idx_name, "columns": [], "non_unique": r.get("Non_unique", 1), "index_type": r.get("Index_type", "BTREE"), } idx_map[idx_name]["columns"].append(r.get("Column_name", "")) indexes = list(idx_map.values()) # Create DDL ddl_sql = "SHOW CREATE TABLE `" + table_name + "`" ddl_qr = await DBConnector.execute_query(instance, database, ddl_sql, user_id=0) if ddl_qr.rows: create_ddl = ddl_qr.rows[0].get("Create Table", "") # Row count cnt_sql = ( "SELECT TABLE_ROWS FROM information_schema.TABLES " "WHERE TABLE_SCHEMA = '" + database + "' " "AND TABLE_NAME = '" + table_name + "'" ) cnt_qr = await DBConnector.execute_query(instance, database, cnt_sql, user_id=0) if cnt_qr.rows: row_count = cnt_qr.rows[0].get("TABLE_ROWS", 0) elif instance.db_type in ("postgres", "postgresql"): # Columns col_sql = ( "SELECT column_name as Field, data_type as Type, " "is_nullable as \"Null\", column_default as \"Default\", " "'' as \"Key\", '' as Extra " "FROM information_schema.columns " "WHERE table_schema = 'public' " "AND table_name = '" + table_name + "' " "ORDER BY ordinal_position" ) col_qr = await DBConnector.execute_query(instance, database, col_sql, user_id=0) columns = col_qr.rows # Indexes idx_sql = ( "SELECT indexname as name, indexdef as definition " "FROM pg_indexes " "WHERE schemaname = 'public' AND tablename = '" + table_name + "'" ) idx_qr = await DBConnector.execute_query(instance, database, idx_sql, user_id=0) indexes = idx_qr.rows # Create DDL — best effort via pg_dump not available; use a placeholder create_ddl = "-- PostgreSQL DDL via pg_indexes:\n" for idx in idx_qr.rows: create_ddl += idx.get("definition", "") + ";\n" # Row count cnt_sql = "SELECT count(*) as cnt FROM \"" + table_name + "\"" try: cnt_qr = await DBConnector.execute_query(instance, database, cnt_sql, user_id=0) if cnt_qr.rows: row_count = cnt_qr.rows[0].get("cnt", 0) except Exception: row_count = -1 elif instance.db_type == "sqlite": # Columns col_sql = "PRAGMA table_info('" + table_name + "')" col_qr = await DBConnector.execute_query(instance, None, col_sql, user_id=0) columns = [] for r in col_qr.rows: columns.append({ "Field": r.get("name", ""), "Type": r.get("type", ""), "Null": "YES" if r.get("notnull", 0) == 0 else "NO", "Key": "PRI" if r.get("pk", 0) == 1 else "", "Default": r.get("dflt_value", ""), "Extra": "", }) # Indexes idx_sql = "PRAGMA index_list('" + table_name + "')" idx_qr = await DBConnector.execute_query(instance, None, idx_sql, user_id=0) indexes = [] for r in idx_qr.rows: indexes.append({ "name": r.get("name", ""), "unique": r.get("unique", 0), }) # Create DDL ddl_sql = ( "SELECT sql FROM sqlite_master WHERE type='table' AND name='" + table_name + "'" ) ddl_qr = await DBConnector.execute_query(instance, None, ddl_sql, user_id=0) if ddl_qr.rows: create_ddl = ddl_qr.rows[0].get("sql", "") # Row count cnt_sql = "SELECT count(*) as cnt FROM '" + table_name + "'" try: cnt_qr = await DBConnector.execute_query(instance, None, cnt_sql, user_id=0) if cnt_qr.rows: row_count = cnt_qr.rows[0].get("cnt", 0) except Exception: row_count = -1 return { "columns": columns, "indexes": indexes, "create_ddl": create_ddl, "row_count": row_count, } async def export_dictionary( session: AsyncSession, instance_id: int, database: str ) -> List[Dict[str, Any]]: """导出整个数据库的表结构信息。""" tables = await list_tables(session, instance_id, database) result = [] for tbl in tables: detail = await get_table_detail(session, instance_id, database, tbl) result.append({ "table_name": tbl, "columns": detail.get("columns", []), "indexes": detail.get("indexes", []), "create_ddl": detail.get("create_ddl", ""), "row_count": detail.get("row_count", 0), }) return result