618 lines
23 KiB
Python
618 lines
23 KiB
Python
from __future__ import annotations
|
||
|
||
from datetime import date, datetime, time, timedelta
|
||
from decimal import Decimal
|
||
import json
|
||
from typing import Any
|
||
|
||
from core.config import (
|
||
MYSQL_CONNECT_TIMEOUT_SECONDS,
|
||
MYSQL_DEFAULT_DATABASE,
|
||
MYSQL_HOST,
|
||
MYSQL_PASSWORD,
|
||
MYSQL_PORT,
|
||
MYSQL_READ_TIMEOUT_SECONDS,
|
||
MYSQL_USER,
|
||
MYSQL_WRITE_TIMEOUT_SECONDS,
|
||
utc_now,
|
||
)
|
||
|
||
|
||
# 系统库默认排在业务库后面，避免主列表被系统库占满。
|
||
SYSTEM_DATABASES = {"information_schema", "performance_schema", "mysql", "sys"}
|
||
|
||
|
||
def mysql_error_payload(error: str) -> dict[str, Any]:
|
||
# MySQL 基础设施页统一返回同一套失败结构，前端不需要额外分支。
|
||
return {
|
||
"ok": False,
|
||
"updatedAt": utc_now(),
|
||
"info": {},
|
||
"connections": {},
|
||
"throughput": {},
|
||
"replication": {"configured": False, "role": "", "readOnly": False, "members": []},
|
||
"summary": {
|
||
"databaseTotal": 0,
|
||
"tableTotal": 0,
|
||
"dataSizeMb": 0.0,
|
||
"indexSizeMb": 0.0,
|
||
"totalSizeMb": 0.0,
|
||
},
|
||
"databases": [],
|
||
"warnings": [],
|
||
"error": error,
|
||
}
|
||
|
||
|
||
def require_pymysql() -> Any:
|
||
try:
|
||
# 运行时再导入驱动，避免缺少依赖时整个服务无法启动。
|
||
import pymysql
|
||
except Exception as exc: # noqa: BLE001
|
||
raise RuntimeError(f"pymysql unavailable: {exc}") from exc
|
||
|
||
# 导入成功后直接返回模块对象，调用方继续拿 DictCursor。
|
||
return pymysql
|
||
|
||
|
||
def build_mysql_connection(database: str | None = None, *, autocommit: bool = True) -> Any:
|
||
# 主机和账号缺失时直接给出明确错误，避免后面连库报一串底层异常。
|
||
missing = [name for name, value in {"MYSQL_HOST": MYSQL_HOST, "MYSQL_USER": MYSQL_USER}.items() if not str(value).strip()]
|
||
if missing:
|
||
raise RuntimeError(f"missing mysql config in .env: {', '.join(missing)}")
|
||
|
||
# 默认库允许为空；表浏览场景会显式选择数据库。
|
||
target_database = str(database or MYSQL_DEFAULT_DATABASE or "").strip() or None
|
||
pymysql = require_pymysql()
|
||
|
||
# 统一走 utf8mb4，避免表结构和数据浏览时乱码。
|
||
return pymysql.connect(
|
||
host=MYSQL_HOST,
|
||
port=MYSQL_PORT,
|
||
user=MYSQL_USER,
|
||
password=MYSQL_PASSWORD,
|
||
database=target_database,
|
||
charset="utf8mb4",
|
||
cursorclass=pymysql.cursors.DictCursor,
|
||
autocommit=autocommit,
|
||
connect_timeout=MYSQL_CONNECT_TIMEOUT_SECONDS,
|
||
read_timeout=MYSQL_READ_TIMEOUT_SECONDS,
|
||
write_timeout=MYSQL_WRITE_TIMEOUT_SECONDS,
|
||
)
|
||
|
||
|
||
def quote_identifier(name: str) -> str:
|
||
# 库、表、列名都统一走反引号包裹，并对内层反引号做转义。
|
||
actual = str(name or "").strip()
|
||
if not actual:
|
||
raise ValueError("identifier is required")
|
||
return f"`{actual.replace('`', '``')}`"
|
||
|
||
|
||
def normalize_database_name(name: str) -> str:
|
||
# 数据库名作为路由参数时必须非空，统一在这里兜底校验。
|
||
actual = str(name or "").strip()
|
||
if not actual:
|
||
raise ValueError("database is required")
|
||
return actual
|
||
|
||
|
||
def normalize_table_name(name: str) -> str:
|
||
# 表名同样必须非空。
|
||
actual = str(name or "").strip()
|
||
if not actual:
|
||
raise ValueError("table is required")
|
||
return actual
|
||
|
||
|
||
def normalize_column_name(name: str) -> str:
|
||
# 列名校验单独抽出来，便于 schema 变更和排序共用。
|
||
actual = str(name or "").strip()
|
||
if not actual:
|
||
raise ValueError("column is required")
|
||
return actual
|
||
|
||
|
||
def is_system_database(name: str) -> bool:
|
||
# 系统库统一小写比较，避免大小写混淆。
|
||
return str(name or "").strip().lower() in SYSTEM_DATABASES
|
||
|
||
|
||
def safe_int(value: Any) -> int:
|
||
try:
|
||
# SHOW STATUS / SHOW TABLE STATUS 这类命令常把数字返回成字符串。
|
||
return int(value or 0)
|
||
except Exception: # noqa: BLE001
|
||
return 0
|
||
|
||
|
||
def safe_float(value: Any) -> float:
|
||
try:
|
||
# 容量类字段统一按浮点数归一，便于后续换算 MiB。
|
||
return float(value or 0.0)
|
||
except Exception: # noqa: BLE001
|
||
return 0.0
|
||
|
||
|
||
def normalize_mysql_value(value: Any) -> Any:
|
||
# Decimal 保留成字符串，避免金额精度被前端浮点化。
|
||
if isinstance(value, Decimal):
|
||
return str(value)
|
||
|
||
# 日期和时间都压成易读字符串。
|
||
if isinstance(value, datetime):
|
||
return value.strftime("%Y-%m-%d %H:%M:%S")
|
||
if isinstance(value, date):
|
||
return value.strftime("%Y-%m-%d")
|
||
if isinstance(value, time):
|
||
return value.strftime("%H:%M:%S")
|
||
if isinstance(value, timedelta):
|
||
return str(value)
|
||
|
||
# 字节内容尽量按 UTF-8 展示；失败时回退成十六进制。
|
||
if isinstance(value, (bytes, bytearray, memoryview)):
|
||
raw = bytes(value)
|
||
try:
|
||
return raw.decode("utf-8")
|
||
except UnicodeDecodeError:
|
||
return f"0x{raw.hex()}"
|
||
|
||
# dict / list 理论上常见于 JSON 列，这里直接转成 JSON 文本方便展示和编辑。
|
||
if isinstance(value, (dict, list)):
|
||
return json.dumps(value, ensure_ascii=False)
|
||
|
||
# 其余标量直接原样返回。
|
||
return value
|
||
|
||
|
||
def normalize_mysql_row(row: dict[str, Any]) -> dict[str, Any]:
|
||
# 一行记录中的每个字段都做 JSON 安全转换。
|
||
return {key: normalize_mysql_value(value) for key, value in row.items()}
|
||
|
||
|
||
def row_first_value(row: Any) -> Any:
|
||
# SHOW DATABASES / SHOW FULL TABLES 的列名会随库名变化，这里统一取第一列。
|
||
if isinstance(row, dict):
|
||
return next(iter(row.values()), None)
|
||
return row
|
||
|
||
|
||
def infer_connected_database(connection: Any) -> str:
|
||
# 当账号拿不到 SHOW DATABASES 时，尽量退到显式配置或当前连接库名。
|
||
for attribute_name in ("db", "database"):
|
||
raw_value = getattr(connection, attribute_name, None)
|
||
if isinstance(raw_value, (bytes, bytearray, memoryview)):
|
||
raw_value = bytes(raw_value).decode("utf-8", errors="ignore")
|
||
actual = str(raw_value or "").strip()
|
||
if actual:
|
||
return actual
|
||
return ""
|
||
|
||
|
||
def list_accessible_database_names(connection: Any) -> list[str]:
|
||
try:
|
||
with connection.cursor() as cursor:
|
||
# 优先使用 SHOW DATABASES，让受限账号也能拿到自己可见的库列表。
|
||
cursor.execute("SHOW DATABASES")
|
||
rows = cursor.fetchall() or []
|
||
except Exception: # noqa: BLE001
|
||
fallback_names = [name for name in {str(MYSQL_DEFAULT_DATABASE or "").strip(), infer_connected_database(connection)} if name]
|
||
if not fallback_names:
|
||
raise
|
||
return sorted(
|
||
fallback_names,
|
||
key=lambda item: (1 if is_system_database(item) else 0, item.lower()),
|
||
)
|
||
|
||
names: list[str] = []
|
||
for item in rows:
|
||
name = str(row_first_value(item) or "").strip()
|
||
if name and name not in names:
|
||
names.append(name)
|
||
return names
|
||
|
||
|
||
def list_accessible_table_names(connection: Any, database: str) -> list[str]:
|
||
target_database = normalize_database_name(database)
|
||
with connection.cursor() as cursor:
|
||
# SHOW FULL TABLES 不依赖 information_schema，可兼容权限更窄的只读账号。
|
||
cursor.execute(f"SHOW FULL TABLES FROM {quote_identifier(target_database)}")
|
||
rows = cursor.fetchall() or []
|
||
|
||
names: list[str] = []
|
||
for item in rows:
|
||
table_name = str(row_first_value(item) or "").strip()
|
||
if table_name and table_name not in names:
|
||
names.append(table_name)
|
||
return names
|
||
|
||
|
||
def show_table_status_rows(connection: Any, database: str) -> list[dict[str, Any]]:
|
||
target_database = normalize_database_name(database)
|
||
|
||
with connection.cursor() as cursor:
|
||
# SHOW TABLE STATUS 能在不查 information_schema 的情况下返回行数和容量估算。
|
||
cursor.execute(f"SHOW TABLE STATUS FROM {quote_identifier(target_database)}")
|
||
return cursor.fetchall() or []
|
||
|
||
|
||
def infer_mysql_data_type(column_type: Any) -> str:
|
||
actual = str(column_type or "").strip().lower()
|
||
if not actual:
|
||
return ""
|
||
return actual.split("(", 1)[0].split()[0]
|
||
|
||
|
||
def sort_database_summaries(items: list[dict[str, Any]]) -> list[dict[str, Any]]:
|
||
# 排序口径保持和 information_schema 查询一致：业务库优先，再按容量倒序。
|
||
return sorted(
|
||
items,
|
||
key=lambda item: (
|
||
1 if item.get("system") else 0,
|
||
-safe_float(item.get("totalSizeMb")),
|
||
str(item.get("name") or "").lower(),
|
||
),
|
||
)
|
||
|
||
|
||
def build_database_summary_item(name: str, table_rows: list[dict[str, Any]]) -> dict[str, Any]:
|
||
data_bytes = sum(safe_float(item.get("Data_length")) for item in table_rows)
|
||
index_bytes = sum(safe_float(item.get("Index_length")) for item in table_rows)
|
||
|
||
return {
|
||
"name": str(name or "").strip(),
|
||
"system": is_system_database(name),
|
||
"tableTotal": len(table_rows),
|
||
"rowEstimate": sum(safe_int(item.get("Rows")) for item in table_rows),
|
||
"dataSizeMb": round(data_bytes / 1024 / 1024, 2),
|
||
"indexSizeMb": round(index_bytes / 1024 / 1024, 2),
|
||
"totalSizeMb": round((data_bytes + index_bytes) / 1024 / 1024, 2),
|
||
}
|
||
|
||
|
||
def list_database_summaries_via_show(connection: Any) -> list[dict[str, Any]]:
|
||
items: list[dict[str, Any]] = []
|
||
|
||
# 逐个库读取 SHOW TABLE STATUS，即使单个库失败也不影响其余库展示。
|
||
for database_name in list_accessible_database_names(connection):
|
||
try:
|
||
table_rows = show_table_status_rows(connection, database_name)
|
||
except Exception: # noqa: BLE001
|
||
table_rows = []
|
||
items.append(build_database_summary_item(database_name, table_rows))
|
||
|
||
return sort_database_summaries(items)
|
||
|
||
|
||
def list_database_summaries(connection: Any) -> list[dict[str, Any]]:
|
||
# 数据库级别汇总统一从 information_schema.tables 读取，能同时拿到表数和容量。
|
||
query = """
|
||
SELECT
|
||
s.SCHEMA_NAME AS schemaName,
|
||
COALESCE(t.tableTotal, 0) AS tableTotal,
|
||
COALESCE(t.dataBytes, 0) AS dataBytes,
|
||
COALESCE(t.indexBytes, 0) AS indexBytes,
|
||
COALESCE(t.rowEstimate, 0) AS rowEstimate
|
||
FROM information_schema.SCHEMATA AS s
|
||
LEFT JOIN (
|
||
SELECT
|
||
table_schema AS schemaName,
|
||
COUNT(*) AS tableTotal,
|
||
SUM(COALESCE(data_length, 0)) AS dataBytes,
|
||
SUM(COALESCE(index_length, 0)) AS indexBytes,
|
||
SUM(COALESCE(table_rows, 0)) AS rowEstimate
|
||
FROM information_schema.TABLES
|
||
GROUP BY table_schema
|
||
) AS t
|
||
ON t.schemaName = s.SCHEMA_NAME
|
||
ORDER BY
|
||
CASE
|
||
WHEN s.SCHEMA_NAME IN ('information_schema', 'performance_schema', 'mysql', 'sys') THEN 1
|
||
ELSE 0
|
||
END,
|
||
COALESCE(t.dataBytes, 0) + COALESCE(t.indexBytes, 0) DESC,
|
||
s.SCHEMA_NAME ASC
|
||
"""
|
||
|
||
try:
|
||
with connection.cursor() as cursor:
|
||
# 直接读取全部库级摘要，前端 sidebar 需要完整列表。
|
||
cursor.execute(query)
|
||
rows = cursor.fetchall() or []
|
||
except Exception: # noqa: BLE001
|
||
# 只读账号常见是 information_schema 可见但全局视图受限，这里退到 SHOW 系列命令。
|
||
return list_database_summaries_via_show(connection)
|
||
|
||
# 把字节数转换成 MiB，并附带是否系统库的标记。
|
||
return sort_database_summaries([
|
||
{
|
||
"name": str(item.get("schemaName") or "").strip(),
|
||
"system": is_system_database(item.get("schemaName")),
|
||
"tableTotal": safe_int(item.get("tableTotal")),
|
||
"rowEstimate": safe_int(item.get("rowEstimate")),
|
||
"dataSizeMb": round(safe_float(item.get("dataBytes")) / 1024 / 1024, 2),
|
||
"indexSizeMb": round(safe_float(item.get("indexBytes")) / 1024 / 1024, 2),
|
||
"totalSizeMb": round(
|
||
(safe_float(item.get("dataBytes")) + safe_float(item.get("indexBytes"))) / 1024 / 1024,
|
||
2,
|
||
),
|
||
}
|
||
for item in rows
|
||
])
|
||
|
||
|
||
def list_table_summaries(connection: Any, database: str) -> list[dict[str, Any]]:
|
||
target_database = normalize_database_name(database)
|
||
|
||
# 表级摘要统一从 information_schema.tables 读取，便于在浏览器里快速切表。
|
||
query = """
|
||
SELECT
|
||
TABLE_NAME AS tableName,
|
||
ENGINE AS engine,
|
||
TABLE_ROWS AS rowEstimate,
|
||
DATA_LENGTH AS dataBytes,
|
||
INDEX_LENGTH AS indexBytes,
|
||
AUTO_INCREMENT AS autoIncrementValue,
|
||
CREATE_TIME AS createTime,
|
||
UPDATE_TIME AS updateTime,
|
||
TABLE_COLLATION AS tableCollation
|
||
FROM information_schema.TABLES
|
||
WHERE TABLE_SCHEMA = %s
|
||
ORDER BY COALESCE(DATA_LENGTH, 0) + COALESCE(INDEX_LENGTH, 0) DESC, TABLE_NAME ASC
|
||
"""
|
||
|
||
try:
|
||
with connection.cursor() as cursor:
|
||
# 同一库内表数量通常有限，这里直接全量读取。
|
||
cursor.execute(query, (target_database,))
|
||
rows = cursor.fetchall() or []
|
||
except Exception: # noqa: BLE001
|
||
rows = show_table_status_rows(connection, target_database)
|
||
return [
|
||
{
|
||
"name": str(item.get("Name") or "").strip(),
|
||
"engine": str(item.get("Engine") or "").strip(),
|
||
"rowEstimate": safe_int(item.get("Rows")),
|
||
"dataSizeMb": round(safe_float(item.get("Data_length")) / 1024 / 1024, 2),
|
||
"indexSizeMb": round(safe_float(item.get("Index_length")) / 1024 / 1024, 2),
|
||
"totalSizeMb": round(
|
||
(safe_float(item.get("Data_length")) + safe_float(item.get("Index_length"))) / 1024 / 1024,
|
||
2,
|
||
),
|
||
"autoIncrement": item.get("Auto_increment"),
|
||
"createTime": normalize_mysql_value(item.get("Create_time")),
|
||
"updateTime": normalize_mysql_value(item.get("Update_time")),
|
||
"collation": str(item.get("Collation") or "").strip(),
|
||
}
|
||
for item in rows
|
||
]
|
||
|
||
return [
|
||
{
|
||
"name": str(item.get("tableName") or "").strip(),
|
||
"engine": str(item.get("engine") or "").strip(),
|
||
"rowEstimate": safe_int(item.get("rowEstimate")),
|
||
"dataSizeMb": round(safe_float(item.get("dataBytes")) / 1024 / 1024, 2),
|
||
"indexSizeMb": round(safe_float(item.get("indexBytes")) / 1024 / 1024, 2),
|
||
"totalSizeMb": round(
|
||
(safe_float(item.get("dataBytes")) + safe_float(item.get("indexBytes"))) / 1024 / 1024,
|
||
2,
|
||
),
|
||
"autoIncrement": item.get("autoIncrementValue"),
|
||
"createTime": normalize_mysql_value(item.get("createTime")),
|
||
"updateTime": normalize_mysql_value(item.get("updateTime")),
|
||
"collation": str(item.get("tableCollation") or "").strip(),
|
||
}
|
||
for item in rows
|
||
]
|
||
|
||
|
||
def list_table_columns(connection: Any, database: str, table: str) -> list[dict[str, Any]]:
|
||
target_database = normalize_database_name(database)
|
||
target_table = normalize_table_name(table)
|
||
|
||
# 列元数据以 ordinal_position 排序，保证浏览器和真实表结构一致。
|
||
query = """
|
||
SELECT
|
||
COLUMN_NAME AS columnName,
|
||
DATA_TYPE AS dataType,
|
||
COLUMN_TYPE AS columnType,
|
||
IS_NULLABLE AS isNullable,
|
||
COLUMN_DEFAULT AS columnDefault,
|
||
COLUMN_KEY AS columnKey,
|
||
EXTRA AS extraValue,
|
||
COLUMN_COMMENT AS columnComment,
|
||
ORDINAL_POSITION AS ordinalPosition
|
||
FROM information_schema.COLUMNS
|
||
WHERE TABLE_SCHEMA = %s AND TABLE_NAME = %s
|
||
ORDER BY ORDINAL_POSITION ASC
|
||
"""
|
||
|
||
try:
|
||
with connection.cursor() as cursor:
|
||
# 表结构操作都依赖列元数据，这里统一一次性拿齐。
|
||
cursor.execute(query, (target_database, target_table))
|
||
rows = cursor.fetchall() or []
|
||
except Exception: # noqa: BLE001
|
||
with connection.cursor() as cursor:
|
||
# SHOW FULL COLUMNS 在受限账号下更稳定，并且字段足够支撑浏览和 DDL。
|
||
cursor.execute(
|
||
f"SHOW FULL COLUMNS FROM {quote_identifier(target_table)} FROM {quote_identifier(target_database)}"
|
||
)
|
||
rows = cursor.fetchall() or []
|
||
return [
|
||
{
|
||
"name": str(item.get("Field") or "").strip(),
|
||
"dataType": infer_mysql_data_type(item.get("Type")),
|
||
"columnType": str(item.get("Type") or "").strip(),
|
||
"nullable": str(item.get("Null") or "").strip().upper() == "YES",
|
||
"default": normalize_mysql_value(item.get("Default")),
|
||
"key": str(item.get("Key") or "").strip(),
|
||
"extra": str(item.get("Extra") or "").strip(),
|
||
"comment": str(item.get("Comment") or "").strip(),
|
||
"ordinalPosition": index,
|
||
}
|
||
for index, item in enumerate(rows, start=1)
|
||
]
|
||
|
||
return [
|
||
{
|
||
"name": str(item.get("columnName") or "").strip(),
|
||
"dataType": str(item.get("dataType") or "").strip(),
|
||
"columnType": str(item.get("columnType") or "").strip(),
|
||
"nullable": str(item.get("isNullable") or "").strip().upper() == "YES",
|
||
"default": normalize_mysql_value(item.get("columnDefault")),
|
||
"key": str(item.get("columnKey") or "").strip(),
|
||
"extra": str(item.get("extraValue") or "").strip(),
|
||
"comment": str(item.get("columnComment") or "").strip(),
|
||
"ordinalPosition": safe_int(item.get("ordinalPosition")),
|
||
}
|
||
for item in rows
|
||
]
|
||
|
||
|
||
def list_table_indexes(connection: Any, database: str, table: str) -> list[dict[str, Any]]:
|
||
target_database = normalize_database_name(database)
|
||
target_table = normalize_table_name(table)
|
||
|
||
# 索引明细先按序号取出，再在 Python 侧聚合成前端更容易消费的结构。
|
||
query = """
|
||
SELECT
|
||
INDEX_NAME AS indexName,
|
||
COLUMN_NAME AS columnName,
|
||
NON_UNIQUE AS nonUnique,
|
||
INDEX_TYPE AS indexType,
|
||
SEQ_IN_INDEX AS seqInIndex,
|
||
SUB_PART AS subPart
|
||
FROM information_schema.STATISTICS
|
||
WHERE TABLE_SCHEMA = %s AND TABLE_NAME = %s
|
||
ORDER BY INDEX_NAME ASC, SEQ_IN_INDEX ASC
|
||
"""
|
||
|
||
try:
|
||
with connection.cursor() as cursor:
|
||
# information_schema.STATISTICS 能完整覆盖 primary / unique / 普通索引。
|
||
cursor.execute(query, (target_database, target_table))
|
||
rows = cursor.fetchall() or []
|
||
except Exception: # noqa: BLE001
|
||
with connection.cursor() as cursor:
|
||
# SHOW INDEX 同样可以完整覆盖索引顺序和类型，并且权限要求更宽松。
|
||
cursor.execute(f"SHOW INDEX FROM {quote_identifier(target_table)} FROM {quote_identifier(target_database)}")
|
||
rows = cursor.fetchall() or []
|
||
|
||
grouped: dict[str, dict[str, Any]] = {}
|
||
for item in rows:
|
||
index_name = str(item.get("indexName") or item.get("Key_name") or "").strip()
|
||
target = grouped.setdefault(
|
||
index_name,
|
||
{
|
||
"name": index_name,
|
||
"primary": index_name == "PRIMARY",
|
||
"unique": safe_int(item.get("nonUnique") if "nonUnique" in item else item.get("Non_unique")) == 0,
|
||
"type": str(item.get("indexType") or item.get("Index_type") or "").strip(),
|
||
"columns": [],
|
||
},
|
||
)
|
||
target["columns"].append(
|
||
{
|
||
"name": str(item.get("columnName") or item.get("Column_name") or "").strip(),
|
||
"subPart": safe_int(item.get("subPart") if "subPart" in item else item.get("Sub_part")) or None,
|
||
"position": safe_int(item.get("seqInIndex") if "seqInIndex" in item else item.get("Seq_in_index")),
|
||
}
|
||
)
|
||
|
||
# 输出时保证 primary 优先，再按名字排序。
|
||
return sorted(
|
||
grouped.values(),
|
||
key=lambda item: (0 if item["primary"] else 1, item["name"].lower()),
|
||
)
|
||
|
||
|
||
def get_preferred_identity_columns(columns: list[dict[str, Any]], indexes: list[dict[str, Any]]) -> dict[str, Any]:
|
||
# 行级更新优先使用主键；没有主键时退到第一个唯一索引；再不行才退到全列比对。
|
||
primary = next((item for item in indexes if item.get("primary")), None)
|
||
if primary:
|
||
return {"mode": "primary", "columns": [column["name"] for column in primary["columns"]]}
|
||
|
||
unique = next((item for item in indexes if item.get("unique")), None)
|
||
if unique:
|
||
return {"mode": "unique", "columns": [column["name"] for column in unique["columns"]]}
|
||
|
||
return {"mode": "all", "columns": [column["name"] for column in columns]}
|
||
|
||
|
||
def build_row_identity(row: dict[str, Any], identity_spec: dict[str, Any]) -> dict[str, Any]:
|
||
# identity 只保留用于匹配的字段，避免前端把整行作为 where 条件带回。
|
||
columns = [str(item or "").strip() for item in identity_spec.get("columns") or [] if str(item or "").strip()]
|
||
return {
|
||
"mode": str(identity_spec.get("mode") or "").strip(),
|
||
"columns": columns,
|
||
"values": {column: normalize_mysql_value(row.get(column)) for column in columns},
|
||
}
|
||
|
||
|
||
def table_exists(connection: Any, database: str, table: str) -> bool:
|
||
target_database = normalize_database_name(database)
|
||
target_table = normalize_table_name(table)
|
||
|
||
# DDL 和浏览接口都需要一个轻量存在性校验，避免后面拼 SQL 才报错。
|
||
try:
|
||
with connection.cursor() as cursor:
|
||
cursor.execute(
|
||
"""
|
||
SELECT 1
|
||
FROM information_schema.TABLES
|
||
WHERE TABLE_SCHEMA = %s AND TABLE_NAME = %s
|
||
LIMIT 1
|
||
""",
|
||
(target_database, target_table),
|
||
)
|
||
return cursor.fetchone() is not None
|
||
except Exception: # noqa: BLE001
|
||
try:
|
||
return target_table in list_accessible_table_names(connection, target_database)
|
||
except Exception: # noqa: BLE001
|
||
return False
|
||
|
||
|
||
def database_exists(connection: Any, database: str) -> bool:
|
||
target_database = normalize_database_name(database)
|
||
|
||
# 库级存在性校验单独抽出来，浏览器切库时可以复用。
|
||
try:
|
||
with connection.cursor() as cursor:
|
||
cursor.execute(
|
||
"""
|
||
SELECT 1
|
||
FROM information_schema.SCHEMATA
|
||
WHERE SCHEMA_NAME = %s
|
||
LIMIT 1
|
||
""",
|
||
(target_database,),
|
||
)
|
||
return cursor.fetchone() is not None
|
||
except Exception: # noqa: BLE001
|
||
try:
|
||
return target_database in list_accessible_database_names(connection)
|
||
except Exception: # noqa: BLE001
|
||
return False
|
||
|
||
|
||
def normalize_sql_text(sql: str) -> str:
|
||
# SQL 工作台当前只允许单条语句，避免一键提交多段脚本。
|
||
text = str(sql or "").strip()
|
||
if not text:
|
||
raise ValueError("sql is required")
|
||
|
||
# 只允许尾部分号，正文里再出现分号就拒绝执行。
|
||
stripped = text.rstrip().rstrip(";").rstrip()
|
||
if ";" in stripped:
|
||
raise ValueError("only single sql statement is allowed")
|
||
return stripped
|
||
|
||
|
||
def sql_statement_kind(sql: str) -> str:
|
||
# 读取首个关键字即可分辨查询类还是变更类语句。
|
||
parts = normalize_sql_text(sql).split(None, 1)
|
||
return parts[0].strip().lower() if parts else ""
|