Commit 6edeb386 authored by Data Governance Dev's avatar Data Governance Dev

feat(web3): 实际数据库连接(参考 web2)

- 后端 core/ 模块自包含(拷贝 web2 的 db_adapter.py / sql_loader.py / models.py + 8 个 SQL 模板),不依赖 web.core
- requirements.txt 加 pymysql / oracledb / dmPython
- routers/db.py 新增 /api/connect/test + /api/connect/schemas + /api/connect/tables 三个端点
- app.py 注册 db_router
- 前端 api/connections.js 封装 3 个调用 + 中文 label→小写 db_type 转换
- TaskModal.vue testConn() / loadSchemas() / loadTables() 改真实后端调用,去掉所有 mock
parent 11d7089d
...@@ -50,11 +50,15 @@ app.add_middleware( ...@@ -50,11 +50,15 @@ app.add_middleware(
) )
from web3.backend.routers.connection_presets import router as connection_presets_router # noqa: E402
from web3.backend.routers.db import router as db_router # noqa: E402
from web3.backend.routers.task_groups import router as task_groups_router # noqa: E402 from web3.backend.routers.task_groups import router as task_groups_router # noqa: E402
from web3.backend.routers.tasks import router as tasks_router # noqa: E402 from web3.backend.routers.tasks import router as tasks_router # noqa: E402
app.include_router(task_groups_router, prefix="/api") app.include_router(task_groups_router, prefix="/api")
app.include_router(tasks_router, prefix="/api") app.include_router(tasks_router, prefix="/api")
logger.info("已注册路由:/api/task-groups, /api/tasks") app.include_router(db_router, prefix="/api")
app.include_router(connection_presets_router, prefix="/api")
logger.info("已注册路由:/api/task-groups, /api/tasks, /api/connect/*, /api/connection-presets")
@app.get("/api/health") @app.get("/api/health")
......
"""数据库适配层(web3 自包含版)
来源:拷贝自 web/core/db_adapter.py(2026-08-20),改为自包含:
- SQL 模板加载器从 web3.backend.core.sql_loader 导入
- 不再依赖 web.core / web.sql
提供 MySQL / 达梦 / Oracle 的统一连接与查询接口,屏蔽方言差异:
- 连接管理
- 字段大小写规范化(达梦默认大写,MySQL 视配置而定)
- 反引号 vs 双引号
- information_schema / 数据字典 查询适配
- 标识符引用(防 SQL 注入)
设计原则:
- 暴露类 PEP-249 cursor-like 接口(execute / fetchall / fetchone)
- 列名统一返回小写 snake_case
- 所有动态拼接的标识符走 quote_ident() 函数
"""
from __future__ import annotations
import logging
import re
import time
from contextlib import contextmanager
from dataclasses import dataclass
from typing import Any, Iterator, Optional
import pymysql
from pymysql.cursors import DictCursor
logger = logging.getLogger(__name__)
# ── 数据类 ────────────────────────────────────────────────
@dataclass
class DBConfig:
"""统一的数据库连接配置"""
db_type: str # "mysql" | "dameng" | "oracle"
host: str
port: int
user: str
password: str
database: str # 数据库名/模式名(Oracle 用作 service_name)
charset: str = "utf8mb4"
connect_timeout: int = 10
# Oracle 专用:Oracle Instant Client 的 lib_dir(oci.dll 所在目录)
# None = 自动搜索 PATH / 注册表 / ORACLE_HOME
# 装了 Instant Client 但不在默认路径时显式指定,如 r"C:\oracle\instantclient_19_8"
# 仅当 thin 模式报 DPY-3010(不支持的 Oracle 版本)时才需要
oracle_client_dir: Optional[str] = None
def to_pymysql_kwargs(self) -> dict:
return {
"host": self.host,
"port": self.port,
"user": self.user,
"password": self.password,
"database": self.database,
"charset": self.charset,
"connect_timeout": self.connect_timeout,
"cursorclass": DictCursor,
}
def to_dm_kwargs(self) -> dict:
"""dmPython 关键字参数映射
关键约束(dmPython 2.5.x,传错会触发 C 层 SystemError、真实异常被吞):
- 主机字段是 ``server``,不是 ``host``(PyMySQL 习惯在这里不适用)
- 不要在 connect 时传 ``schema``:达梦 schema = 用户,登录用户默认就在自己的 schema;
如果 user 与 database 不一致,连上后单独 ``SET SCHEMA "xxx"`` 即可
- 不要在 connect 时传 ``charset``:dmPython 默认 GB18030(与 disql 一致),
部分驱动版本对 local_code 取值校验严苛(CODE:-70023),让驱动默认最稳
- 不要在 connect 时传 ``connect_timeout``:dmPython 2.5.x 对这个 kwarg 的
支持视 wheel 而异,不传最安全(超时由 OS TCP 重传决定)
参考:scripts/_probe_dm.py + 同环境已验证可连的 export_all_tables.py 工作脚本
"""
return {
"user": self.user,
"password": self.password,
"server": self.host,
"port": self.port,
}
def to_oracle_kwargs(self) -> dict:
"""oracledb 关键字参数映射(thin 模式,无需 Instant Client)
- database 字段在 Oracle 里是 service_name(如 ORCLPDB1),不是 schema 名
(Oracle schema = user,治理任务通常以「一个 service 扫一个 schema」的视角使用)
- 超时参数叫 tcp_connect_timeout(oracledb 1.4.x 起就用这个名)
单位秒,默认未指定;旧 1.x 风格的 connect_timeout 在 1.4.2+ 已被移除,传了会报
TypeError: connect() got an unexpected keyword argument 'connect_timeout'
- 不传 charset:thin 模式从 OS NLS_LANG / 系统设置取,避免与 NLS 字符集冲突
"""
return {
"user": self.user,
"password": self.password,
"host": self.host,
"port": self.port,
"service_name": self.database,
"tcp_connect_timeout": float(self.connect_timeout),
}
# ── 标识符引用 ────────────────────────────────────────────
_IDENT_RE = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*$")
def quote_ident(name: str, db_type: str) -> str:
"""安全引用 SQL 标识符(表名 / 字段名)。
- MySQL 默认用反引号
- 达梦 / Oracle 用双引号(ANSI 标准)
- 仅允许字母数字下划线开头,其余字符拒绝以防注入
"""
if not _IDENT_RE.match(name):
raise ValueError(f"非法标识符: {name!r}")
if db_type == "mysql":
return f"`{name}`"
return f'"{name}"'
def quote_value(value: Any) -> str:
"""把 Python 值转成 SQL 字面量。仅用于已知安全的字典/枚举值。"""
if value is None:
return "NULL"
if isinstance(value, bool):
return "1" if value else "0"
if isinstance(value, (int, float)):
return str(value)
# 字符串转义(仅供受信任字面量使用,业务表名/字段名走 quote_ident)
s = str(value).replace("'", "''")
return f"'{s}'"
# ── 列名规范化 ────────────────────────────────────────────
def normalize_column_name(name: str) -> str:
"""统一列名为小写 + 保持其余内容"""
return name.lower() if isinstance(name, str) else name
def _rows_as_dicts(rows, col_names: list[str]) -> list[dict]:
"""把不同驱动的查询结果统一转成 [{col_lower: value}, ...]
- PyMySQL 用 DictCursor → 单行是 dict,直接 .items()
- dmPython / cx_Oracle / oracledb → 单行是 tuple,用 cur.description 拿列名
"""
if not rows:
return []
sample = rows[0]
if isinstance(sample, dict):
# PyMySQL DictCursor 路径
return [{normalize_column_name(k): v for k, v in row.items()} for row in rows]
# tuple / list —— 用 col_names 兜底转 dict
return [{normalize_column_name(c): v for c, v in zip(col_names, row)} for row in rows]
def normalize_rows(rows: list[dict] | None) -> list[dict]:
"""兼容老调用:单行 dict 的 rows 直接规范化列名"""
if not rows:
return []
return [{normalize_column_name(k): v for k, v in row.items()} for row in rows]
# ── 信息模式查询(通过 SQL 加载器,无内联 SQL) ──────────
# 所有 SQL 已抽到 web3/backend/core/sql_templates/ 目录下的 .sql 模板
# schema 一律走绑定参数(MySQL %s / 达梦 ?),不拼进 SQL 文本
def _render_info_columns_sql(dialect: str) -> str:
from web3.backend.core.sql_loader import get_sql_loader
return get_sql_loader().render("info_schema/list_columns", dialect=dialect)
def _render_info_tables_sql(dialect: str) -> str:
from web3.backend.core.sql_loader import get_sql_loader
return get_sql_loader().render("info_schema/list_tables", dialect=dialect)
def _render_health_sql(dialect: str) -> str:
from web3.backend.core.sql_loader import get_sql_loader
return get_sql_loader().render("health/check_connection", dialect=dialect)
# ── Oracle Instant Client 加载(thick 模式) ──────────────
# oracledb thin 模式只支持 Oracle 11.2+。老版本(10g / 9i)必须切 thick,
# 即加载本地 Oracle Instant Client 的 oci.dll(Windows)/ libclntsh.so(Linux)。
#
# 调用 oracledb.init_oracle_client() 是「进程级」操作:
# - 同一进程多次调用:除首次外都是 no-op
# - 失败后无法在同一进程回退到 thin(要重启进程)
#
# 自动检测策略:默认搜索 PATH / 注册表 / ORACLE_HOME,
# 也可以通过 DBConfig.oracle_client_dir 显式指定 lib_dir。
_oracle_client_initialized: bool = False
def _ensure_oracle_client(lib_dir: Optional[str] = None) -> str:
"""尝试加载 Oracle Instant Client,返回实际生效的连接模式
Returns:
"thick" Instant Client 加载成功(支持老 Oracle)
"thin" 未找到 Instant Client(仅支持 Oracle 11.2+,依赖 oracledb 自带协议栈)
Raises:
不抛异常 —— 即便加载失败也安全返回 "thin",让上层用 thin 试连,
然后把 DPY-3010 报给用户看,提示装 Instant Client。
"""
global _oracle_client_initialized
if _oracle_client_initialized:
return "thick"
try:
import oracledb
if lib_dir:
oracledb.init_oracle_client(lib_dir=lib_dir)
else:
oracledb.init_oracle_client() # 默认搜索 PATH / 注册表 / ORACLE_HOME
_oracle_client_initialized = True
logger.info(f"[DB] Oracle Instant Client 已加载(thick 模式){f'lib_dir={lib_dir}' if lib_dir else '(默认搜索路径)'}")
return "thick"
except Exception as e:
err_str = str(e)
# DPI-1047 + oci.dll 存在:最常见原因是缺 VC++ Redistributable
# (Instant Client 12.x 是用 MSVC 2013 编译的,依赖 MSVCR120.dll;
# 装 vcredist 2015-2022 不够,必须装 VC++ 2013 Redist)
if "DPI-1047" in err_str:
import os
oci_in_lib_dir = False
if lib_dir:
oci_in_lib_dir = os.path.exists(os.path.join(lib_dir, "oci.dll"))
extra_hint = ""
if lib_dir and oci_in_lib_dir:
# 自动扫描 oci.dll 的导入表,列出它依赖的运行时 DLL
# (用 PE 文件格式解析导入表;不依赖 dumpbin / PowerShell)
missing_runtime_hint = ""
try:
import struct
oci_path = os.path.join(lib_dir, "oci.dll")
pe_bytes = open(oci_path, "rb").read()
pe_off = struct.unpack_from("<I", pe_bytes, 0x3C)[0]
opt_off = pe_off + 24
is_64 = (struct.unpack_from("<H", pe_bytes, opt_off)[0] == 0x20B)
data_dir_off = opt_off + (112 if is_64 else 96)
import_rva = struct.unpack_from("<I", pe_bytes, data_dir_off + 8)[0]
# Section table → RVA ↔ file offset
num_sections = struct.unpack_from("<H", pe_bytes, pe_off + 6)[0]
size_opt_hdr = struct.unpack_from("<H", pe_bytes, pe_off + 20)[0]
sec_table_off = pe_off + 24 + size_opt_hdr
def _rva_to_off(rva):
for i in range(num_sections):
so = sec_table_off + i * 40
vaddr = struct.unpack_from("<I", pe_bytes, so + 12)[0]
vsize = struct.unpack_from("<I", pe_bytes, so + 8)[0]
raw_ptr = struct.unpack_from("<I", pe_bytes, so + 20)[0]
if vaddr <= rva < vaddr + vsize:
return raw_ptr + (rva - vaddr)
return None
imports = []
tbl_off = _rva_to_off(import_rva)
while tbl_off is not None:
entry = pe_bytes[tbl_off:tbl_off + 20]
if len(entry) < 20: break
_, _, _, name_rva, _ = struct.unpack("<IIIII", entry)
if name_rva == 0: break
n_off = _rva_to_off(name_rva)
if n_off is None: break
end = pe_bytes.find(b"\x00", n_off)
imports.append(pe_bytes[n_off:end].decode("ascii", errors="replace"))
tbl_off += 20
# 过滤掉系统 DLL,只列可能缺 vcredist 的运行时
system_dlls = {"KERNEL32.DLL", "ADVAPI32.DLL", "USER32.DLL", "GDI32.DLL",
"WS2_32.DLL", "CRYPT32.DLL", "SECUR32.DLL", "OLE32.DLL",
"OLEAUT32.DLL", "SHELL32.DLL", "SHLWAPI.DLL", "WININET.DLL",
"WINHTTP.DLL", "IPHLPAPI.DLL", "NETAPI32.DLL", "PSAPI.DLL",
"MSWSOCK.DLL", "WSOCK32.DLL", "NORMALIZ.DLL", "API-MS-WIN-CRT-*.DLL"}
runtime_deps = [
d for d in imports
if not any(d.upper().startswith(p.upper()) for p in system_dlls)
]
if runtime_deps:
missing_runtime_hint = (
f"\n 诊断:oci.dll 依赖的运行时 DLL = {runtime_deps}\n"
" (如果是 MSVCR120.dll → 装 Visual C++ 2013 Redistributable (x64):"
"https://www.microsoft.com/en-us/download/details.aspx?id=40784)\n"
" (如果是 MSVCP140.dll / VCRUNTIME140.dll → 装 vc_redist.x64.exe:"
"https://aka.ms/vs/17/release/vc_redist.x64.exe)\n"
" 注意:Instant Client 12.x 是 MSVC 2013 编译的(_MSC_VER=1800),"
"装 vc_redist 2015-2022 不够,必须装 2013!"
)
except Exception as parse_e:
missing_runtime_hint = f"\n (PE 解析失败: {parse_e})"
extra_hint = (
"\n 诊断:lib_dir 下 oci.dll 存在,但仍报 DPI-1047 → "
"最常见原因是缺 VC++ Redistributable"
+ missing_runtime_hint
)
elif lib_dir:
extra_hint = (
f"\n 诊断:lib_dir={lib_dir} 下找不到 oci.dll → 路径不对\n"
" 修复:lib_dir 应该填到含 oci.dll 的那一级(不是上一级)"
)
logger.info(
f"[DB] 未找到 Oracle Instant Client(DPI-1047: {e})→ 使用 thin 模式{extra_hint}"
)
else:
logger.info(
f"[DB] 未找到 Oracle Instant Client({type(e).__name__}: {e})→ 使用 thin 模式,"
f"仅支持 Oracle 11.2+。如连老版本 Oracle 请装 Instant Client"
f"(https://www.oracle.com/database/technologies/instant-client/)"
)
return "thin"
# ── 字符串类型集合 ────────────────────────────────────────
TEXT_TYPES = {
# MySQL
"varchar", "char", "text", "longtext", "mediumtext", "tinytext", "enum", "set",
# 达梦
"varchar2", "nvarchar2", "char", "text", "clob", "longvarchar",
# Oracle(结构与达梦高度重合,新增 nclob / long)
"varchar2", "nvarchar2", "char", "clob", "nclob", "long",
}
# ── 数值类型集合(自定义规则 v2 用:> / < / = 比较符) ──
NUMERIC_TYPES = {
# MySQL
"int", "integer", "smallint", "tinyint", "bigint", "mediumint",
"decimal", "numeric", "float", "double", "real", "year",
# 达梦 / Oracle
"number", "decimal", "numeric", "float", "double", "real", "binary_float", "binary_double",
"money", "smallmoney", "dec",
}
# ── 日期/时间类型集合(自定义规则 v2 用:早于 / 晚于 比较符) ──
DATE_TYPES = {
# MySQL
"date", "datetime", "timestamp", "time", "year",
# 达梦 / Oracle / SQL Server
"date", "datetime", "timestamp", "time",
"smalldatetime", "datetime2", "datetimeoffset", "interval",
"timestamp with time zone", "timestamp with local time zone",
}
# ── 连接管理 ──────────────────────────────────────────────
class DBConnection:
"""统一的数据库连接封装。
用法(SQL 一律走 web3/backend/core/sql_templates/ 模板,不要在 Python 里拼 SQL):
with DBConnection(cfg) as db:
rows = db.list_columns("mydb") # schema 走绑定参数
# 需要自定义模板时,值用 params 传,标识符才用 ${x | quote}
from web3.backend.core.sql_loader import get_sql_loader
sql = get_sql_loader().render("verify/count_table_rows",
dialect=cfg.db_type, table="t_project")
n = db.fetch_scalar(sql)
"""
def __init__(self, cfg: DBConfig):
self.cfg = cfg
self._conn = None
self._driver = self._select_driver(cfg.db_type)
@staticmethod
def _select_driver(db_type: str):
if db_type == "mysql":
return "pymysql"
if db_type == "dameng":
try:
import dmPython # noqa: F401
except ImportError as e:
raise RuntimeError(
"未安装 dmPython,请执行: pip install dmPython\n"
"(达梦官方驱动需先安装达梦客户端)"
) from e
return "dmPython"
if db_type == "oracle":
try:
import oracledb # noqa: F401
except ImportError as e:
raise RuntimeError(
"未安装 oracledb,请执行: pip install oracledb\n"
"(oracledb thin 模式无需 Oracle Instant Client)"
) from e
return "oracledb"
raise ValueError(f"不支持的数据库类型: {db_type!r}(目前支持 mysql / dameng / oracle)")
def __enter__(self):
logger.info(
f"[DB] 打开连接: {self.cfg.db_type}://{self.cfg.user}@{self.cfg.host}:"
f"{self.cfg.port}/{self.cfg.database} (driver={self._driver})"
)
self._connect_started = time.monotonic()
try:
if self._driver == "pymysql":
self._conn = pymysql.connect(**self.cfg.to_pymysql_kwargs())
elif self._driver == "dmPython":
import dmPython
# dmPython 是 C 扩展,底层异常经常被 SystemError 吃掉。
try:
self._conn = dmPython.connect(**self.cfg.to_dm_kwargs())
except SystemError as e:
# dmPython 真实错误经常是以下几种之一:
# - dmPython 与本机达梦客户端(libdmdpi.so / dmdpi.dll)版本不匹配
# - 字符集不支持(达梦常见 GB18030 / UTF-8)
# - 网络/端口/服务名错(达梦错误被 C 层吃掉)
extra_hints = (
"常见原因:\n"
" 1) dmPython 与本机达梦客户端版本不匹配(dmPython 包版本 vs libdmdpi.so / dmdpi.dll)\n"
" 2) charset 不被达梦支持(达梦常见 GB18030 / UTF-8,MySQL 风格的 utf8mb4 / utf8 不一定可用)\n"
" 3) 主机/端口/服务名/账号密码错(真实 ORA-like 错误被 C 层吃了)\n"
"调试: 在终端单独跑 scripts/_probe_dm.py 看真实错误"
)
logger.error(
f"[DB] dmPython 连接失败(C 扩展吞了真实异常): {e}\n"
f" 入参: {self.cfg.to_dm_kwargs()}\n{extra_hints}"
)
raise
else: # oracledb
import oracledb
# 自动检测 Instant Client:能找到就切 thick 模式(兼容老 Oracle)
mode = _ensure_oracle_client(self.cfg.oracle_client_dir)
self._oracle_mode = mode # 记下来给日志用
logger.debug(f"[DB] Oracle 连接模式: {mode}")
self._conn = oracledb.connect(**self.cfg.to_oracle_kwargs())
except Exception as e:
# Oracle 连接失败时多打一些诊断信息:oracledb 版本 + connect() 合法 kwarg 列表
# 避免「TypeError: got an unexpected keyword argument 'xxx'」把根因藏起来
logger.error(f"[DB] 连接失败: {type(e).__name__}: {e}")
if self._driver == "oracledb":
err_str = str(e)
# ── 典型错误 1:thin 模式不支持老 Oracle(11.2 以下)──
# DPY-3010: connections to this database server version are not supported
# 修复:装 Oracle Instant Client 让 _ensure_oracle_client() 切 thick
if "DPY-3010" in err_str:
logger.error(
"[DB] Oracle 诊断: 当前 Oracle 服务端版本低于 11.2(或 thin 模式不支持)\n"
" 修复步骤:\n"
" 1. 下载 Instant Client Basic 包(约 100MB):\n"
" https://www.oracle.com/database/technologies/instant-client/downloads.html\n"
" 2. 解压到本地(Windows 示例)D:\\oracle\\instantclient_19_8\n"
" 3. 前端「Oracle Client」输入框填该路径,重启 web 服务后再测\n"
" (或加到 PATH / 设置 ORACLE_HOME 让 _ensure_oracle_client() 自动找到)"
)
# ── 典型错误 2:Instant Client 缺失(DPI-1047)──
# 通常和 DPY-3010 一起出现,因为 _ensure_oracle_client() 没装上 → 走 thin → DPY-3010
elif "DPI-1047" in err_str:
logger.error(
"[DB] Oracle 诊断: Instant Client 缺失(DPI-1047)\n"
" 修复:同 DPY-3010 的步骤 1~3(下载解压并填到前端 Oracle Client 输入框)"
)
# ── 其他错误:照旧打 inspect 信息 ──
else:
try:
import inspect
import oracledb as _odb
valid_kwargs = list(inspect.signature(_odb.connect).parameters.keys())
logger.error(
f"[DB] Oracle 诊断: oracledb.__version__={_odb.__version__}, "
f"connect() 合法 kwarg={valid_kwargs}"
)
logger.error(f"[DB] Oracle 诊断: 实际传入 kwarg={self.cfg.to_oracle_kwargs()}")
except Exception as diag_e:
logger.error(f"[DB] Oracle 诊断打印失败: {diag_e}")
raise
elapsed_ms = (time.monotonic() - self._connect_started) * 1000
logger.info(f"[DB] 连接已建立(耗时 {elapsed_ms:.0f}ms)")
return self
def __exit__(self, exc_type, exc, tb):
if self._conn is not None:
try:
self._conn.close()
logger.debug("[DB] 连接已关闭")
except Exception as e:
logger.warning(f"[DB] 关闭连接异常: {e}")
# ── 查询方法(自动规范化列名) ──
def _cursor(self):
if self._conn is None:
raise RuntimeError("DBConnection 未打开,请用 with 语句")
return self._conn.cursor()
def fetchall(self, sql: str, params: tuple | None = None) -> list[dict]:
cur = self._cursor()
started = time.monotonic()
try:
cur.execute(sql, params or ())
rows = cur.fetchall()
# 提前取列名:dmPython/oracledb 返回 tuple,需要 description 兜底
col_names = [d[0] for d in (cur.description or [])]
elapsed_ms = (time.monotonic() - started) * 1000
logger.debug(f"[SQL] fetchall {len(rows)} rows ({elapsed_ms:.1f}ms): {sql[:120]}...")
return _rows_as_dicts(rows, col_names)
finally:
cur.close()
def fetchone(self, sql: str, params: tuple | None = None) -> Optional[dict]:
cur = self._cursor()
started = time.monotonic()
try:
cur.execute(sql, params or ())
row = cur.fetchone()
col_names = [d[0] for d in (cur.description or [])]
elapsed_ms = (time.monotonic() - started) * 1000
logger.debug(f"[SQL] fetchone ({elapsed_ms:.1f}ms): {sql[:120]}...")
if row is None:
return None
if isinstance(row, dict):
return {normalize_column_name(k): v for k, v in row.items()}
return {normalize_column_name(c): v for c, v in zip(col_names, row)}
finally:
cur.close()
def fetch_scalar(self, sql: str, params: tuple | None = None) -> Any:
row = self.fetchone(sql, params)
if not row:
return None
return next(iter(row.values()))
def execute(self, sql: str, params: tuple | None = None) -> int:
cur = self._cursor()
started = time.monotonic()
try:
n = cur.execute(sql, params or ())
elapsed_ms = (time.monotonic() - started) * 1000
logger.debug(f"[SQL] execute {n} rows ({elapsed_ms:.1f}ms): {sql[:120]}...")
return n
finally:
cur.close()
# ── 信息模式便捷方法(schema 走绑定参数,不拼接) ──────
def list_columns(self, schema: str) -> list[dict]:
"""返回指定 schema 下所有列的元数据
SQL 通过 web3/backend/core/sql_templates/info_schema/list_columns.<dialect>.sql 加载,
schema 作为绑定参数传给 cursor,由驱动层负责转义,杜绝 SQL 注入。
"""
sql = _render_info_columns_sql(self.cfg.db_type)
return self.fetchall(sql, (schema,))
def list_tables(self, schema: str) -> list[dict]:
"""返回指定 schema 下所有表的元数据(schema 走绑定参数)"""
sql = _render_info_tables_sql(self.cfg.db_type)
return self.fetchall(sql, (schema,))
def _fetchall_named(self, sql: str, params: dict) -> list[dict]:
"""保留方法以兼容达梦风格的命名占位符(暂未使用,预留)"""
cur = self._cursor()
try:
cur.execute(sql, params)
return normalize_rows(cur.fetchall())
finally:
cur.close()
# ── 上下文管理器快捷方式 ──────────────────────────────────
@contextmanager
def open_db(cfg: DBConfig) -> Iterator[DBConnection]:
with DBConnection(cfg) as conn:
yield conn
def test_connection(cfg: DBConfig) -> tuple[bool, str]:
"""测试连接是否可用,返回 (ok, message)"""
logger.info(
f"[DB] 测试连接: {cfg.db_type}://{cfg.user}@{cfg.host}:{cfg.port}/{cfg.database}"
)
try:
with DBConnection(cfg) as db:
sql = _render_health_sql(cfg.db_type)
v = db.fetch_scalar(sql)
if v == 1:
logger.info("[DB] 测试连接成功")
return True, f"连接成功({cfg.db_type} @ {cfg.host}:{cfg.port}/{cfg.database})"
logger.warning(f"[DB] 测试连接返回非预期值: {v}")
return False, "连接成功但查询返回非预期值"
except Exception as e:
logger.error(f"[DB] 测试连接失败: {type(e).__name__}: {e}")
return False, f"连接失败: {e}"
\ No newline at end of file
"""web3 数据库连接 Pydantic 模型
来源:精简自 web/core/models.py(2026-08-20),只保留数据库连接所需的三个类。
"""
from __future__ import annotations
from typing import Optional
from pydantic import BaseModel, Field
# ── 请求 ────────────────────────────────────────────────
class ConnectRequest(BaseModel):
"""统一的数据库连接请求基类(测试连接 + 列出 schema/表 共用)"""
db_type: str = Field(..., description="mysql / dameng / oracle")
host: str
port: int = 3306
user: str
password: str
database: str
charset: str = "utf8mb4"
connect_timeout: int = 10
# Oracle 专用:Oracle Instant Client 的 lib_dir(oci.dll 所在目录)
oracle_client_dir: Optional[str] = None
class TestConnectionRequest(ConnectRequest):
"""只测连通性,不启动任务;前端不需要填 tables(会被忽略)"""
tables: list[str] = Field(default_factory=list)
# ── 响应 ────────────────────────────────────────────────
class TestConnectionResponse(BaseModel):
ok: bool
message: str
db_type: Optional[str] = None
database_name: Optional[str] = None # 改名为避免与 BaseModel.schema() 冲突
tables_count: Optional[int] = None
table_summary: list[dict] = Field(default_factory=list)
data_dictionary: list[dict] = Field(default_factory=list)
\ No newline at end of file
"""SQL 模板加载器(web3 自包含版)
来源:精简自 web/sql/loader.py(2026-08-20)。
变化点:
- 模板根目录改为 web3/backend/core/sql_templates/
- 移除 `${var | quote}` filter(web3 用法不需要;quote 由调用方在 Python 侧处理)
- 移除对 web.core.db_adapter.quote_ident 的依赖
把所有 SQL 语句集中到 web3/backend/core/sql_templates/ 目录下,Python 代码不再直接拼 SQL。
模板使用 ${var} 占位符(避免与 SQL 自身的 {} 冲突)。
支持的占位符:
- ${var_name} 简单字符串替换
- ${list | join:","} 列表拼接(默认分隔符 ",")
- ${list | join:" AND "} 自定义分隔符
Dialect 加载规则:
- 优先加载 <name>.<dialect>.sql
- 回退到 <name>.sql(通用版本)
- 都没有则报错
示例:
loader = get_sql_loader()
sql = loader.render("info_schema/list_tables", dialect="mysql", schema="mydb")
"""
from __future__ import annotations
import logging
import re
from pathlib import Path
from threading import Lock
from typing import Any
logger = logging.getLogger(__name__)
# ── 占位符语法 ────────────────────────────────────────────
# 支持:${var} / ${var | filter} / ${var | filter:"sep"}
# 不支持嵌套 ${${a}} 这种(SQL 模板里不需要)
_PLACEHOLDER_RE = re.compile(
r"\$\{\s*([A-Za-z_][A-Za-z0-9_]*)\s*"
r"(?:\|\s*([A-Za-z_]+)\s*(?::\s*\"([^\"]*)\")?\s*)?\}"
)
def _render_placeholders(template: str, params: dict) -> str:
"""把 ${var} / ${var | filter} 替换为 params 里的值
内置 filter:
- join[:"<sep>"]: 把列表按 sep 拼接,默认 ","
- upper: 转大写
- lower: 转小写
"""
def _replace(m: re.Match) -> str:
var_name = m.group(1)
filt = m.group(2)
sep = m.group(3) # filter 的字符串参数
if var_name not in params:
raise KeyError(
f"SQL 模板参数缺失: {var_name!r}(提供: {list(params.keys())})"
)
value = params[var_name]
# 应用 filter
if filt == "join":
if not isinstance(value, (list, tuple)):
raise TypeError(f"join filter 需要 list/tuple,{var_name} 是 {type(value).__name__}")
return sep.join(str(v) for v in value)
elif filt == "upper":
return str(value).upper()
elif filt == "lower":
return str(value).lower()
elif filt is None:
return str(value)
else:
raise ValueError(f"未知 filter: {filt!r}({var_name})")
return _PLACEHOLDER_RE.sub(_replace, template)
# ── 加载器 ────────────────────────────────────────────────
class SQLLoader:
"""SQL 模板加载器(单例)"""
def __init__(self, base_dir: Path | None = None):
# web3/backend/core/sql_loader.py → web3/backend/core/sql_templates/ 是模板根目录
self.base_dir = base_dir or (Path(__file__).resolve().parent / "sql_templates")
self._cache: dict[str, str] = {}
self._lock = Lock()
# ── 加载 ──
def _find_template_file(self, name: str, dialect: str) -> Path:
"""优先找 .<dialect>.sql,回退 .sql"""
dsp = self.base_dir / f"{name}.{dialect}.sql"
if dsp.exists():
return dsp
common = self.base_dir / f"{name}.sql"
if common.exists():
return common
raise FileNotFoundError(
f"SQL 模板不存在: {name}(dialect={dialect});"
f"已查找: {dsp.relative_to(self.base_dir)} 和 {common.relative_to(self.base_dir)}"
)
def _load_text(self, name: str, dialect: str) -> str:
cache_key = f"{dialect}::{name}"
with self._lock:
if cache_key in self._cache:
return self._cache[cache_key]
path = self._find_template_file(name, dialect)
text = path.read_text(encoding="utf-8")
self._cache[cache_key] = text
logger.debug(f"加载 SQL 模板: {path.relative_to(self.base_dir)}")
return text
# ── 渲染 ──
def render(self, name: str, dialect: str = "mysql", **params: Any) -> str:
"""渲染 SQL 模板
Args:
name: 模板路径(不含 .sql 后缀),如 "info_schema/list_tables"
dialect: 数据库方言 "mysql" / "dameng" / "oracle"
**params: 占位符参数
Returns:
渲染后的 SQL 字符串
"""
text = self._load_text(name, dialect)
try:
return _render_placeholders(text, params)
except KeyError as e:
raise KeyError(f"SQL 模板 {name!r} (dialect={dialect}): {e}") from None
# ── 清缓存(热加载调试用) ──
def clear_cache(self) -> None:
with self._lock:
self._cache.clear()
# ── 全局单例 ──────────────────────────────────────────────
_instance: SQLLoader | None = None
def get_sql_loader() -> SQLLoader:
"""获取 SQL 加载器单例"""
global _instance
if _instance is None:
_instance = SQLLoader()
return _instance
def reset_sql_loader() -> None:
"""重置(测试用)"""
global _instance
_instance = None
\ No newline at end of file
-- ============================================================================
-- 健康检查:确认连接可用,返回固定值 1
-- 调用方:web/core/db_adapter.py → test_connection()
-- 方言:Oracle
--
-- Oracle 严格遵循 SQL 标准 —— SELECT 必须有 FROM 子句,
-- 因此不能像 MySQL 那样写 SELECT 1,必须 FROM dual。
-- dual 是 Oracle 内置的单行单列表,恒定返回 1 行。
-- ============================================================================
SELECT 1 FROM dual
\ No newline at end of file
-- ============================================================================
-- 健康检查:确认连接可用,返回固定值 1
-- 调用方:web/core/db_adapter.py → test_connection()
-- 跨方言通用
-- ============================================================================
SELECT 1
\ No newline at end of file
-- ============================================================================
-- 列出指定 schema 下所有列的元数据 (达梦方言)
-- 调用方:web/core/db_adapter.py → list_columns()
-- 参数:schema 数据库/模式名(绑定参数,由调用方通过 params 传入,勿拼接)
-- 说明:
-- - 达梦的 ALL_TAB_COLUMNS.COMMENTS 列名与 MySQL 的 COLUMN_COMMENT 不同
-- - 达梦的 OWNER 通常为大写
-- - 列注释不在 ALL_TAB_COLUMNS 里,要 LEFT JOIN ALL_COL_COMMENTS(Oracle 同源)
-- ============================================================================
SELECT
c.TABLE_NAME AS table_name,
c.COLUMN_NAME AS column_name,
c.COLUMN_ID AS ordinal_position,
c.DATA_TYPE || CASE WHEN c.DATA_LENGTH IS NOT NULL THEN '(' || c.DATA_LENGTH || ')' END AS column_type,
c.DATA_TYPE AS data_type,
c.DATA_LENGTH AS char_max_length,
c.DATA_PRECISION AS numeric_precision,
c.DATA_SCALE AS numeric_scale,
c.NULLABLE AS is_nullable,
c.DATA_DEFAULT AS column_default,
cc.COMMENTS AS column_comment,
NULL AS extra
FROM ALL_TAB_COLUMNS c
LEFT JOIN ALL_COL_COMMENTS cc
ON cc.OWNER = c.OWNER
AND cc.TABLE_NAME = c.TABLE_NAME
AND cc.COLUMN_NAME = c.COLUMN_NAME
WHERE c.OWNER = ?
ORDER BY c.TABLE_NAME, c.COLUMN_ID
-- ============================================================================
-- 列出指定 schema 下所有列的元数据 (MySQL 方言)
-- 调用方:web/core/db_adapter.py → list_columns()
-- 参数:schema 数据库/模式名(绑定参数,由调用方通过 params 传入,勿拼接)
-- 注意:本文件不得出现其它百分号字面量,否则会被驱动误认为占位符
-- 返回字段:table_name, column_name, ordinal_position, column_type,
-- data_type, char_max_length, numeric_precision, numeric_scale,
-- is_nullable, column_default, column_comment, extra
-- ============================================================================
SELECT
c.TABLE_NAME AS table_name,
c.COLUMN_NAME AS column_name,
c.ORDINAL_POSITION AS ordinal_position,
c.COLUMN_TYPE AS column_type,
c.DATA_TYPE AS data_type,
c.CHARACTER_MAXIMUM_LENGTH AS char_max_length,
c.NUMERIC_PRECISION AS numeric_precision,
c.NUMERIC_SCALE AS numeric_scale,
c.IS_NULLABLE AS is_nullable,
c.COLUMN_DEFAULT AS column_default,
c.COLUMN_COMMENT AS column_comment,
c.EXTRA AS extra
FROM INFORMATION_SCHEMA.COLUMNS c
WHERE c.TABLE_SCHEMA = %s
ORDER BY c.TABLE_NAME, c.ORDINAL_POSITION
\ No newline at end of file
-- ============================================================================
-- 列出指定 schema 下所有列的元数据 (Oracle 方言)
-- 调用方:web/core/db_adapter.py → list_columns()
-- 参数:schema 用户/模式名(绑定参数,调用方通过 params 传入,勿拼接)
-- 说明:
-- - Oracle 没有 INFORMATION_SCHEMA,使用 ALL_TAB_COLUMNS(结构与达梦高度一致)
-- - Oracle 的 OWNER 通常为大写;调用方传入的小写 schema 也用 UPPER() 兼容
-- - 占位符用 :1(oracledb 接收 named/numbered binding)
-- - DATA_TYPE 在字符类型上拼 DATA_LENGTH;NUMBER 类型没有 DATA_LENGTH,单独显示
-- - is_nullable 列在 Oracle 中是 NULLABLE(取值 'Y'/'N'),与达梦相同
-- - 列注释不在 ALL_TAB_COLUMNS 里,要 LEFT JOIN ALL_COL_COMMENTS
-- ============================================================================
SELECT
c.TABLE_NAME AS table_name,
c.COLUMN_NAME AS column_name,
c.COLUMN_ID AS ordinal_position,
CASE
WHEN c.DATA_TYPE IN ('VARCHAR2','NVARCHAR2','CHAR','RAW')
THEN c.DATA_TYPE || '(' || c.DATA_LENGTH || ')'
ELSE c.DATA_TYPE
END AS column_type,
c.DATA_TYPE AS data_type,
c.DATA_LENGTH AS char_max_length,
c.DATA_PRECISION AS numeric_precision,
c.DATA_SCALE AS numeric_scale,
c.NULLABLE AS is_nullable,
c.DATA_DEFAULT AS column_default,
cc.COMMENTS AS column_comment,
NULL AS extra
FROM ALL_TAB_COLUMNS c
LEFT JOIN ALL_COL_COMMENTS cc
ON cc.OWNER = c.OWNER
AND cc.TABLE_NAME = c.TABLE_NAME
AND cc.COLUMN_NAME = c.COLUMN_NAME
WHERE c.OWNER = UPPER(:1)
ORDER BY c.TABLE_NAME, c.COLUMN_ID
-- ============================================================================
-- 列出指定 schema 下所有表 (达梦方言)
-- 调用方:web/core/db_adapter.py → list_tables()
-- 参数:schema 数据库/模式名(绑定参数,由调用方通过 params 传入,勿拼接)
-- 说明:
-- - 达梦 ALL_TABLES 没有 BYTES 列(Oracle 才有),DATA_LENGTH 给 NULL
-- - 达梦 ALL_TABLES 也没有 COMMENTS 列,表注释存在 ALL_TAB_COMMENTS 里,
-- 要 LEFT JOIN 出来(与 ALL_COL_COMMENTS / ALL_TAB_COLUMNS 同源结构)
-- - 达梦 ALL_TABLES 也没有 CREATED / LAST_DDL 列(时间戳一概没有),
-- 真要看得 LEFT JOIN ALL_OBJECTS(按 OBJECT_TYPE='TABLE' 过滤),
-- 字段治理主线不用时间戳,先给 NULL 兜底
-- ============================================================================
SELECT t.TABLE_NAME AS table_name,
'BASE TABLE' AS table_type,
NULL AS engine,
t.NUM_ROWS AS table_rows,
NULL AS data_length,
0 AS index_length,
tc.COMMENTS AS table_comment,
NULL AS create_time,
NULL AS update_time
FROM ALL_TABLES t
LEFT JOIN ALL_TAB_COMMENTS tc
ON tc.OWNER = t.OWNER
AND tc.TABLE_NAME = t.TABLE_NAME
WHERE t.OWNER = UPPER(?)
ORDER BY t.TABLE_NAME
\ No newline at end of file
-- ============================================================================
-- 列出指定 schema 下所有表 (MySQL 方言)
-- 调用方:web/core/db_adapter.py → list_tables()
-- 参数:schema 数据库/模式名(绑定参数,由调用方通过 params 传入,勿拼接)
-- 注意:本文件不得出现其它百分号字面量,否则会被驱动误认为占位符
-- ============================================================================
SELECT TABLE_NAME, TABLE_TYPE, ENGINE, TABLE_ROWS,
DATA_LENGTH, INDEX_LENGTH, TABLE_COMMENT,
CREATE_TIME, UPDATE_TIME
FROM INFORMATION_SCHEMA.TABLES
WHERE TABLE_SCHEMA = %s
ORDER BY TABLE_NAME
\ No newline at end of file
-- ============================================================================
-- 列出指定 schema 下所有表 (Oracle 方言)
-- 调用方:web/core/db_adapter.py → list_tables()
-- 参数:schema 用户/模式名(绑定参数,调用方通过 params 传入,勿拼接)
-- 说明:
-- - Oracle 没有 INFORMATION_SCHEMA,使用 ALL_TABLES(结构与达梦高度一致)
-- - Oracle ALL_TABLES 没有 COMMENTS 列,表注释在 ALL_TAB_COMMENTS 里,
-- LEFT JOIN 出来(与达梦同源——同一类问题同一个修法)
-- - 2026-08-13:实测 11g (11.2.0.4) ALL_TABLES 实际可用列清单(用户跑 DESC ALL_TABLES 确认):
-- OWNER / TABLE_NAME / TABLESPACE_NAME / CLUSTER_NAME / IOT_NAME / STATUS /
-- PCT_FREE / PCT_USED / INI_TRANS / MAX_TRANS /
-- INITIAL_EXTENT / NEXT_EXTENT / MIN_EXTENTS / MAX_EXTENTS / PCT_INCREASE /
-- FREELISTS / FREELIST_GROUPS / LOGGING / BACKED_UP /
-- NUM_ROWS / BLOCKS / EMPTY_BLOCKS / AVG_SPACE / CHAIN_CNT / AVG_ROW_LEN /
-- AVG_SPACE_FREELIST_BLOCKS / NUM_FREELIST_BLOCKS
-- - 11g ALL_TABLES **没有** BYTES / CREATED / LAST_DDL_TIME 列(这三个都是 12c+ 加的),
-- 所以 data_length 用 BLOCKS * 8192 估算(Oracle 默认 block size 8KB;空表为 0,未分配 segment),
-- create_time / update_time 取 NULL。若以后接 12c+,新增 list_tables.oracle.12c.sql
-- 即可(sql_loader 自动优先选方言版本,不影响 11g)。
-- ============================================================================
SELECT t.TABLE_NAME AS table_name,
'BASE TABLE' AS table_type,
NULL AS engine,
t.NUM_ROWS AS table_rows,
t.BLOCKS * 8192 AS data_length, -- 11g 没 BYTES,用 BLOCKS×8K 估算
0 AS index_length,
tc.COMMENTS AS table_comment,
NULL AS create_time, -- 11g ALL_TABLES 没有 CREATED(12c+)
NULL AS update_time -- 11g ALL_TABLES 没有 LAST_DDL_TIME(12c+)
FROM ALL_TABLES t
LEFT JOIN ALL_TAB_COMMENTS tc
ON tc.OWNER = t.OWNER
AND tc.TABLE_NAME = t.TABLE_NAME
WHERE t.OWNER = UPPER(:1)
ORDER BY t.TABLE_NAME
fastapi>=0.110 fastapi>=0.110
uvicorn[standard]>=0.27 uvicorn[standard]>=0.27
sqlalchemy>=2.0 sqlalchemy>=2.0
pydantic>=2.5 pydantic>=2.5
\ No newline at end of file # 数据库驱动(web3/backend/core/db_adapter.py 需要;dmPython 装不上不影响启动,只影响连达梦)
pymysql>=1.1
oracledb>=1.4
dmPython>=2.5
\ No newline at end of file
"""web3 数据库连接 API
端点:
POST /api/connect/test 测试连接(MySQL / Oracle / 达梦)
POST /api/connect/schemas 列出可用的 schema / database / user
POST /api/connect/tables 列出指定 schema 下的所有表
实现:
- 复用 web3/backend/core/db_adapter(自包含,不依赖 web.core)
- db_type 兼容前端中文 label("MySQL"/"Oracle"/"达梦 DM"),内部统一小写
- 错误用 logger.exception 拿全栈,响应 message 给前端弹窗
- 错误分类提示(WinError 10061 / timeout / Access denied)写到 logs/app.log
"""
from __future__ import annotations
import time
from typing import Any
from fastapi import APIRouter
from web3.backend.core.db_adapter import DBConfig, DBConnection
from web3.backend.core.sql_loader import get_sql_loader
from web3.backend.core.models import TestConnectionRequest, TestConnectionResponse
from web3.backend._logging import get_logger
router = APIRouter(prefix="/connect", tags=["database"])
logger = get_logger("backend.routers.db")
# ── db_type 映射 ──────────────────────────────────────────
# 前端 form.dbType 是中文 label("MySQL"/"Oracle"/"达梦 DM"),内部统一小写(mysql/oracle/dameng)
_LABEL_TO_DB_TYPE = {
"MySQL": "mysql",
"Oracle": "oracle",
"达梦 DM": "dameng",
}
def _normalize_db_type(db_type: str) -> str:
"""把前端传入的 db_type 归一化成小写内部值"""
return _LABEL_TO_DB_TYPE.get(db_type, db_type.lower())
def _build_db_config(req: TestConnectionRequest) -> DBConfig:
"""把 TestConnectionRequest 转换成 DBConfig"""
return DBConfig(
db_type=_normalize_db_type(req.db_type),
host=req.host,
port=req.port,
user=req.user,
password=req.password,
database=req.database,
charset=req.charset,
connect_timeout=req.connect_timeout,
oracle_client_dir=req.oracle_client_dir,
)
# ── 工具函数 ─────────────────────────────────────────────
def _mask_password(cfg: DBConfig) -> str:
"""DBConfig 脱敏输出(用于日志)"""
return (
f"db_type={cfg.db_type} "
f"host={cfg.host}:{cfg.port} "
f"user={cfg.user} "
f"password=*** "
f"database={cfg.database}"
)
def _log_failure_hint(msg: str) -> None:
"""根据错误 message 分类写日志(给运维看,不影响响应)"""
if "WinError 10061" in msg or "Connection refused" in msg:
logger.warning(
" 💡 诊断: 目标端口未监听(连接被拒)。可能原因:\n"
" 1) 主机/端口写错;\n"
" 2) 数据库服务没启;\n"
" 3) 防火墙拦截;\n"
" 4) 数据库不在当前网络可达段"
)
elif "timed out" in msg.lower() or "timeout" in msg.lower():
logger.warning(
" 💡 诊断: 连接超时。可能原因:\n"
" 1) 主机不可达(IP 错 / 不在同一网段);\n"
" 2) 防火墙丢包;\n"
" 3) 数据库服务卡死但不拒绝连接"
)
elif "Access denied" in msg or "1045" in msg:
logger.warning(" 💡 诊断: 用户名或密码错")
elif "Unknown database" in msg or "1049" in msg:
logger.warning(" 💡 诊断: 数据库名不存在")
# ── 端点 ─────────────────────────────────────────────────
@router.post("/test", response_model=TestConnectionResponse, summary="测试数据库连接")
async def connect_test(req: TestConnectionRequest):
"""复用 web3.backend.core.db_adapter,验证 db_type/host/port/user/password/database
返回 TestConnectionResponse(ok, message, db_type, database_name)
"""
cfg = _build_db_config(req)
logger.info("─" * 60)
logger.info(f"POST /api/connect/test {_mask_password(cfg)}")
started = time.monotonic()
try:
with DBConnection(cfg) as db:
sql = get_sql_loader().render("health/check_connection", dialect=cfg.db_type)
v = db.fetch_scalar(sql)
elapsed_ms = (time.monotonic() - started) * 1000
if v == 1:
msg = f"连接成功({cfg.db_type} @ {cfg.host}:{cfg.port}/{cfg.database},耗时 {elapsed_ms:.0f}ms)"
logger.info(f"✅ {msg}")
logger.info("─" * 60)
return TestConnectionResponse(
ok=True,
message=msg,
db_type=cfg.db_type,
database_name=cfg.database,
)
msg = "连接成功但查询返回非预期值"
logger.warning(f"❌ {msg}(耗时 {elapsed_ms:.0f}ms)")
logger.info("─" * 60)
return TestConnectionResponse(ok=False, message=msg, db_type=cfg.db_type)
except Exception as e:
elapsed_ms = (time.monotonic() - started) * 1000
msg = f"连接失败: {type(e).__name__}: {e}"
# logger.exception 自动带全栈
logger.exception(f"❌ 测试连接抛出异常(耗时 {elapsed_ms:.0f}ms)")
_log_failure_hint(str(e))
logger.info("─" * 60)
return TestConnectionResponse(ok=False, message=msg, db_type=cfg.db_type)
@router.post("/schemas", summary="列出可用的 schema / database / user")
async def list_schemas(req: TestConnectionRequest):
"""根据 db_type 列出可连接的 schema 列表
- MySQL = SHOW DATABASES
- 达梦 = SELECT DISTINCT OWNER FROM ALL_TABLES(排除系统用户)
- Oracle = SELECT username FROM all_users(排除系统用户)
"""
cfg = _build_db_config(req)
logger.info("─" * 60)
logger.info(f"POST /api/connect/schemas {_mask_password(cfg)}")
try:
with DBConnection(cfg) as db:
rows: list[dict[str, Any]] = []
if cfg.db_type == "mysql":
rows = db.fetchall("SHOW DATABASES")
schemas = [r.get("Database", r.get("database", "")) for r in rows]
elif cfg.db_type == "dameng":
rows = db.fetchall(
"SELECT DISTINCT OWNER AS schema_name FROM ALL_TABLES "
"WHERE OWNER NOT IN ('SYS','SYSTEM','SYSAUDITOR','CTISYS') "
"ORDER BY OWNER"
)
schemas = [r["schema_name"] for r in rows]
elif cfg.db_type == "oracle":
rows = db.fetchall(
"SELECT username AS schema_name FROM all_users "
"WHERE username NOT IN ('SYS','SYSTEM','OUTLN','DIP',"
"'ORACLE_OCM','XDB','ANONYMOUS','CTXSYS','MDSYS','OLAPSYS',"
"'ORDDATA','APEX_030200','APEX_PUBLIC_USER','FLOWS_FILES') "
"ORDER BY username"
)
schemas = [r["schema_name"] for r in rows]
else:
msg = f"不支持的 db_type: {cfg.db_type}"
logger.warning(f"❌ {msg}")
logger.info("─" * 60)
return {"ok": False, "message": msg, "schemas": []}
logger.info(f"✅ 返回 {len(schemas)} 个 schema")
logger.info("─" * 60)
return {"ok": True, "schemas": schemas, "count": len(schemas)}
except Exception as e:
msg = f"获取 schema 列表失败: {type(e).__name__}: {e}"
logger.exception(f"❌ {msg}")
_log_failure_hint(str(e))
logger.info("─" * 60)
return {"ok": False, "message": msg, "schemas": []}
@router.post("/tables", summary="列出指定 schema 下的所有表")
async def list_tables(req: TestConnectionRequest):
"""复用 db_adapter.list_tables(),走 info_schema/list_tables.<dialect>.sql 模板"""
cfg = _build_db_config(req)
logger.info("─" * 60)
logger.info(f"POST /api/connect/tables {_mask_password(cfg)} schema={req.database!r}")
try:
with DBConnection(cfg) as db:
rows = db.list_tables(req.database)
logger.info(f"✅ 返回 {len(rows)} 张表")
logger.info("─" * 60)
return {"ok": True, "tables": rows, "count": len(rows)}
except Exception as e:
msg = f"获取表列表失败: {type(e).__name__}: {e}"
logger.exception(f"❌ {msg}")
_log_failure_hint(str(e))
logger.info("─" * 60)
return {"ok": False, "message": msg, "tables": []}
\ No newline at end of file
/**
* 数据库连接 API(2026-08-20 接入真实后端)
*
* 前端 form.dbType 是中文 label("MySQL"/"Oracle"/"达梦 DM"),
* 后端 /api/connect/* 端点统一接收小写(mysql/oracle/dameng),这里做归一化。
*/
import { http } from './client'
const DB_TYPE_MAP = {
'MySQL': 'mysql',
'Oracle': 'oracle',
'达梦 DM': 'dameng',
}
/**
* 把前端 form 转成后端 TestConnectionRequest 字段
* - dbType 中文 label → 内部小写
* - port 字符串 → number
* - database 用 form.db(前端字段名是 db,后端字段是 database)
*/
function buildPayload(form) {
return {
db_type: DB_TYPE_MAP[form.dbType] || form.dbType,
host: form.host,
port: Number(form.port) || 0,
user: form.user,
password: form.password,
database: form.db,
charset: 'utf8mb4',
connect_timeout: 10,
}
}
/**
* 测试数据库连接
* @returns { ok, message, db_type, database_name, tables_count, ... }
*/
export function testConnection(form) {
return http.post('/connect/test', buildPayload(form))
}
/**
* 列出可用的 schema / database / user
* @returns { ok, schemas: string[], count, message? }
*/
export function listSchemas(form) {
return http.post('/connect/schemas', buildPayload(form))
}
/**
* 列出指定 schema 下的所有表
* @returns { ok, tables: [{table_name, table_comment, ...}], count, message? }
*/
export function listTables(form) {
return http.post('/connect/tables', buildPayload(form))
}
\ No newline at end of file
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment