720 lines
28 KiB
Python
720 lines
28 KiB
Python
from __future__ import annotations
|
||
|
||
import json
|
||
from typing import Any
|
||
|
||
from core.config import MYSQL_DEFAULT_DATABASE, MYSQL_DEFAULT_PAGE_SIZE, MYSQL_MAX_PAGE_SIZE, MYSQL_QUERY_MAX_ROWS, utc_now
|
||
from .mysql_common import (
|
||
build_mysql_connection,
|
||
build_row_identity,
|
||
database_exists,
|
||
get_preferred_identity_columns,
|
||
list_database_summaries,
|
||
list_table_columns,
|
||
list_table_indexes,
|
||
list_table_summaries,
|
||
normalize_column_name,
|
||
normalize_database_name,
|
||
normalize_mysql_row,
|
||
normalize_sql_text,
|
||
normalize_table_name,
|
||
quote_identifier,
|
||
sql_statement_kind,
|
||
table_exists,
|
||
)
|
||
from .mysql_metrics import clear_mysql_metrics_cache, get_mysql_overview_payload
|
||
|
||
|
||
def browser_error_payload(error: str) -> dict[str, Any]:
|
||
# 浏览器和 SQL 工作台都统一返回这套最小错误结构。
|
||
return {"ok": False, "updatedAt": utc_now(), "error": error}
|
||
|
||
|
||
def clamp_page_size(value: Any) -> int:
|
||
try:
|
||
# 前端没传时回到默认 100；过大时强制截断，避免一次拖太多数据。
|
||
return max(1, min(int(value or MYSQL_DEFAULT_PAGE_SIZE), MYSQL_MAX_PAGE_SIZE))
|
||
except Exception: # noqa: BLE001
|
||
return MYSQL_DEFAULT_PAGE_SIZE
|
||
|
||
|
||
def clamp_page_number(value: Any) -> int:
|
||
try:
|
||
# 页码最小固定为 1。
|
||
return max(1, int(value or 1))
|
||
except Exception: # noqa: BLE001
|
||
return 1
|
||
|
||
|
||
def normalize_sql_fragment(value: Any, *, field_name: str) -> str:
|
||
# DDL 片段允许空格和括号，但不允许多语句。
|
||
actual = str(value or "").strip()
|
||
if not actual:
|
||
raise ValueError(f"{field_name} is required")
|
||
if ";" in actual:
|
||
raise ValueError(f"{field_name} must be a single fragment")
|
||
return actual
|
||
|
||
|
||
def get_mysql_databases_payload() -> dict[str, Any]:
|
||
# 概览里已经做了缓存，这里直接复用，避免重复查 information_schema。
|
||
overview = get_mysql_overview_payload()
|
||
if not overview.get("ok"):
|
||
return {
|
||
"ok": False,
|
||
"updatedAt": overview.get("updatedAt"),
|
||
"selectedDatabase": "",
|
||
"items": [],
|
||
"error": overview.get("error") or "mysql unavailable",
|
||
}
|
||
|
||
items = list(overview.get("databases") or [])
|
||
business_items = [item for item in items if not item.get("system")]
|
||
preferred = str(MYSQL_DEFAULT_DATABASE or "").strip()
|
||
|
||
# 默认优先选 .env 指定库；没有时选第一个业务库，再退到第一个可见库。
|
||
selected_database = (
|
||
preferred
|
||
if preferred and any(item.get("name") == preferred for item in items)
|
||
else (business_items[0]["name"] if business_items else (items[0]["name"] if items else ""))
|
||
)
|
||
|
||
return {
|
||
"ok": True,
|
||
"updatedAt": overview.get("updatedAt"),
|
||
"selectedDatabase": selected_database,
|
||
"items": items,
|
||
"error": "",
|
||
}
|
||
|
||
|
||
def get_mysql_tables_payload(database: str) -> dict[str, Any]:
|
||
target_database = normalize_database_name(database)
|
||
|
||
try:
|
||
with build_mysql_connection(autocommit=True) as connection:
|
||
if not database_exists(connection, target_database):
|
||
raise ValueError(f"database not found: {target_database}")
|
||
items = list_table_summaries(connection, target_database)
|
||
except Exception as exc: # noqa: BLE001
|
||
return {
|
||
"ok": False,
|
||
"updatedAt": utc_now(),
|
||
"database": target_database,
|
||
"items": [],
|
||
"error": str(exc),
|
||
}
|
||
|
||
return {
|
||
"ok": True,
|
||
"updatedAt": utc_now(),
|
||
"database": target_database,
|
||
"selectedTable": items[0]["name"] if items else "",
|
||
"items": items,
|
||
"error": "",
|
||
}
|
||
|
||
|
||
def get_mysql_table_schema_payload(database: str, table: str) -> dict[str, Any]:
|
||
target_database = normalize_database_name(database)
|
||
target_table = normalize_table_name(table)
|
||
|
||
try:
|
||
with build_mysql_connection(target_database, autocommit=True) as connection:
|
||
if not table_exists(connection, target_database, target_table):
|
||
raise ValueError(f"table not found: {target_database}.{target_table}")
|
||
|
||
tables = list_table_summaries(connection, target_database)
|
||
table_summary = next((item for item in tables if item.get("name") == target_table), None)
|
||
columns = list_table_columns(connection, target_database, target_table)
|
||
indexes = list_table_indexes(connection, target_database, target_table)
|
||
|
||
with connection.cursor() as cursor:
|
||
# SHOW CREATE TABLE 能拿到最完整的 DDL，方便页面直接展示和复制。
|
||
cursor.execute(f"SHOW CREATE TABLE {quote_identifier(target_table)}")
|
||
create_row = cursor.fetchone() or {}
|
||
except Exception as exc: # noqa: BLE001
|
||
return {
|
||
"ok": False,
|
||
"updatedAt": utc_now(),
|
||
"database": target_database,
|
||
"table": target_table,
|
||
"columns": [],
|
||
"indexes": [],
|
||
"createTable": "",
|
||
"error": str(exc),
|
||
}
|
||
|
||
return {
|
||
"ok": True,
|
||
"updatedAt": utc_now(),
|
||
"database": target_database,
|
||
"table": target_table,
|
||
"tableSummary": table_summary or {},
|
||
"columns": columns,
|
||
"indexes": indexes,
|
||
"identity": get_preferred_identity_columns(columns, indexes),
|
||
"createTable": str(create_row.get("Create Table") or "").strip(),
|
||
"error": "",
|
||
}
|
||
|
||
|
||
def build_filter_clause(
|
||
columns: list[dict[str, Any]],
|
||
keyword: str,
|
||
filters: list[dict[str, Any]],
|
||
) -> tuple[str, list[Any]]:
|
||
column_names = {column["name"] for column in columns}
|
||
clauses: list[str] = []
|
||
params: list[Any] = []
|
||
normalized_keyword = str(keyword or "").strip()
|
||
|
||
# 关键字搜索默认对当前表全部列做模糊匹配，适合值班快速扫数据。
|
||
if normalized_keyword:
|
||
search_clauses: list[str] = []
|
||
for column in columns:
|
||
search_clauses.append(f"CAST({quote_identifier(column['name'])} AS CHAR) LIKE %s")
|
||
params.append(f"%{normalized_keyword}%")
|
||
if search_clauses:
|
||
clauses.append(f"({' OR '.join(search_clauses)})")
|
||
|
||
# 附加过滤器支持前端后续扩展成多条件。
|
||
for item in filters:
|
||
column_name = normalize_column_name(item.get("column"))
|
||
if column_name not in column_names:
|
||
raise ValueError(f"unknown filter column: {column_name}")
|
||
|
||
operator = str(item.get("operator") or "eq").strip().lower()
|
||
sql_column = quote_identifier(column_name)
|
||
value = item.get("value")
|
||
|
||
if operator == "eq":
|
||
if value is None:
|
||
clauses.append(f"{sql_column} IS NULL")
|
||
else:
|
||
clauses.append(f"{sql_column} = %s")
|
||
params.append(value)
|
||
continue
|
||
if operator == "ne":
|
||
if value is None:
|
||
clauses.append(f"{sql_column} IS NOT NULL")
|
||
else:
|
||
clauses.append(f"{sql_column} <> %s")
|
||
params.append(value)
|
||
continue
|
||
if operator == "gt":
|
||
clauses.append(f"{sql_column} > %s")
|
||
params.append(value)
|
||
continue
|
||
if operator == "ge":
|
||
clauses.append(f"{sql_column} >= %s")
|
||
params.append(value)
|
||
continue
|
||
if operator == "lt":
|
||
clauses.append(f"{sql_column} < %s")
|
||
params.append(value)
|
||
continue
|
||
if operator == "le":
|
||
clauses.append(f"{sql_column} <= %s")
|
||
params.append(value)
|
||
continue
|
||
if operator == "in":
|
||
if not isinstance(value, list) or not value:
|
||
raise ValueError("in filter requires a non-empty array value")
|
||
placeholders = ", ".join(["%s"] * len(value))
|
||
clauses.append(f"{sql_column} IN ({placeholders})")
|
||
params.extend(value)
|
||
continue
|
||
if operator == "not_in":
|
||
if not isinstance(value, list) or not value:
|
||
raise ValueError("not_in filter requires a non-empty array value")
|
||
placeholders = ", ".join(["%s"] * len(value))
|
||
clauses.append(f"{sql_column} NOT IN ({placeholders})")
|
||
params.extend(value)
|
||
continue
|
||
if operator == "between":
|
||
if not isinstance(value, list) or len(value) != 2:
|
||
raise ValueError("between filter requires a two-item array value")
|
||
clauses.append(f"{sql_column} BETWEEN %s AND %s")
|
||
params.extend(value)
|
||
continue
|
||
if operator == "not_between":
|
||
if not isinstance(value, list) or len(value) != 2:
|
||
raise ValueError("not_between filter requires a two-item array value")
|
||
clauses.append(f"{sql_column} NOT BETWEEN %s AND %s")
|
||
params.extend(value)
|
||
continue
|
||
if operator == "like":
|
||
clauses.append(f"CAST({sql_column} AS CHAR) LIKE %s")
|
||
params.append(f"%{value}%")
|
||
continue
|
||
if operator == "not_like":
|
||
clauses.append(f"CAST({sql_column} AS CHAR) NOT LIKE %s")
|
||
params.append(f"%{value}%")
|
||
continue
|
||
if operator == "is_null":
|
||
clauses.append(f"{sql_column} IS NULL")
|
||
continue
|
||
if operator == "is_not_null":
|
||
clauses.append(f"{sql_column} IS NOT NULL")
|
||
continue
|
||
|
||
raise ValueError(f"unsupported filter operator: {operator}")
|
||
|
||
if not clauses:
|
||
return "", params
|
||
return f"WHERE {' AND '.join(clauses)}", params
|
||
|
||
|
||
def query_mysql_table_rows_payload(payload: dict[str, Any]) -> dict[str, Any]:
|
||
target_database = normalize_database_name(payload.get("database"))
|
||
target_table = normalize_table_name(payload.get("table"))
|
||
page = clamp_page_number(payload.get("page"))
|
||
page_size = clamp_page_size(payload.get("pageSize"))
|
||
keyword = str(payload.get("keyword") or "").strip()
|
||
filters = payload.get("filters") if isinstance(payload.get("filters"), list) else []
|
||
|
||
try:
|
||
with build_mysql_connection(target_database, autocommit=True) as connection:
|
||
if not table_exists(connection, target_database, target_table):
|
||
raise ValueError(f"table not found: {target_database}.{target_table}")
|
||
|
||
columns = list_table_columns(connection, target_database, target_table)
|
||
indexes = list_table_indexes(connection, target_database, target_table)
|
||
identity_spec = get_preferred_identity_columns(columns, indexes)
|
||
column_names = {column["name"] for column in columns}
|
||
|
||
requested_sort = str(payload.get("orderBy") or "").strip()
|
||
order_by = requested_sort if requested_sort in column_names else (identity_spec["columns"][0] if identity_spec["columns"] else columns[0]["name"])
|
||
order_direction = "DESC" if str(payload.get("orderDirection") or "").strip().lower() == "desc" else "ASC"
|
||
|
||
where_clause, params = build_filter_clause(columns, keyword, filters)
|
||
offset = (page - 1) * page_size
|
||
quoted_table = quote_identifier(target_table)
|
||
select_columns = ", ".join(quote_identifier(column["name"]) for column in columns)
|
||
|
||
with connection.cursor() as cursor:
|
||
# 先统计总量，分页器和页数都依赖这个值。
|
||
cursor.execute(
|
||
f"SELECT COUNT(*) AS totalRows FROM {quoted_table} {where_clause}",
|
||
tuple(params),
|
||
)
|
||
total_rows = int((cursor.fetchone() or {}).get("totalRows") or 0)
|
||
|
||
# 行浏览默认按指定列排序，避免翻页时顺序漂移。
|
||
cursor.execute(
|
||
f"""
|
||
SELECT {select_columns}
|
||
FROM {quoted_table}
|
||
{where_clause}
|
||
ORDER BY {quote_identifier(order_by)} {order_direction}
|
||
LIMIT %s OFFSET %s
|
||
""",
|
||
tuple([*params, page_size, offset]),
|
||
)
|
||
rows = cursor.fetchall() or []
|
||
except Exception as exc: # noqa: BLE001
|
||
return {
|
||
"ok": False,
|
||
"updatedAt": utc_now(),
|
||
"database": target_database,
|
||
"table": target_table,
|
||
"rows": [],
|
||
"totalRows": 0,
|
||
"error": str(exc),
|
||
}
|
||
|
||
# 每一行都带 identity，后续 update / delete 不需要前端自己猜 where 条件。
|
||
normalized_rows = []
|
||
for row in rows:
|
||
normalized_rows.append(
|
||
{
|
||
"values": normalize_mysql_row(row),
|
||
"identity": build_row_identity(row, identity_spec),
|
||
}
|
||
)
|
||
|
||
return {
|
||
"ok": True,
|
||
"updatedAt": utc_now(),
|
||
"database": target_database,
|
||
"table": target_table,
|
||
"page": page,
|
||
"pageSize": page_size,
|
||
"totalRows": total_rows,
|
||
"totalPages": max(1, (total_rows + page_size - 1) // page_size) if total_rows > 0 else 1,
|
||
"keyword": keyword,
|
||
"orderBy": order_by,
|
||
"orderDirection": order_direction.lower(),
|
||
"columns": columns,
|
||
"indexes": indexes,
|
||
"identity": identity_spec,
|
||
"rows": normalized_rows,
|
||
"error": "",
|
||
}
|
||
|
||
|
||
def normalize_write_values(values: Any, columns: list[dict[str, Any]]) -> list[tuple[str, Any]]:
|
||
if not isinstance(values, dict):
|
||
raise ValueError("values must be an object")
|
||
|
||
writable_columns = {
|
||
column["name"]: column
|
||
for column in columns
|
||
if "generated" not in str(column.get("extra") or "").strip().lower()
|
||
}
|
||
normalized: list[tuple[str, Any]] = []
|
||
|
||
# 只允许写入真实存在的列，避免页面或调用方传进来脏字段。
|
||
for raw_name, raw_value in values.items():
|
||
column_name = normalize_column_name(raw_name)
|
||
if column_name not in writable_columns:
|
||
raise ValueError(f"unknown writable column: {column_name}")
|
||
|
||
# JSON 对象和数组转成字符串后再交给 MySQL，自定义 SQL 控制台仍可手写更复杂格式。
|
||
if isinstance(raw_value, (dict, list)):
|
||
normalized.append((column_name, json.dumps(raw_value, ensure_ascii=False)))
|
||
else:
|
||
normalized.append((column_name, raw_value))
|
||
|
||
return normalized
|
||
|
||
|
||
def build_identity_clause(identity: dict[str, Any], columns: list[dict[str, Any]]) -> tuple[str, list[Any]]:
|
||
if not isinstance(identity, dict):
|
||
raise ValueError("identity is required")
|
||
|
||
allowed_columns = {column["name"] for column in columns}
|
||
identity_columns = [normalize_column_name(item) for item in identity.get("columns") or []]
|
||
identity_values = identity.get("values") if isinstance(identity.get("values"), dict) else {}
|
||
|
||
if not identity_columns:
|
||
raise ValueError("identity columns are required")
|
||
|
||
clauses: list[str] = []
|
||
params: list[Any] = []
|
||
|
||
# where 条件只按 identity 指定列构造，避免把整行都塞进 where。
|
||
for column_name in identity_columns:
|
||
if column_name not in allowed_columns:
|
||
raise ValueError(f"unknown identity column: {column_name}")
|
||
if column_name not in identity_values:
|
||
raise ValueError(f"missing identity value: {column_name}")
|
||
|
||
value = identity_values.get(column_name)
|
||
quoted = quote_identifier(column_name)
|
||
if value is None:
|
||
clauses.append(f"{quoted} IS NULL")
|
||
else:
|
||
clauses.append(f"{quoted} = %s")
|
||
params.append(value)
|
||
|
||
return " AND ".join(clauses), params
|
||
|
||
|
||
def insert_mysql_row_payload(payload: dict[str, Any]) -> dict[str, Any]:
|
||
target_database = normalize_database_name(payload.get("database"))
|
||
target_table = normalize_table_name(payload.get("table"))
|
||
|
||
try:
|
||
with build_mysql_connection(target_database, autocommit=False) as connection:
|
||
columns = list_table_columns(connection, target_database, target_table)
|
||
pairs = normalize_write_values(payload.get("values"), columns)
|
||
if not pairs:
|
||
raise ValueError("at least one column value is required")
|
||
|
||
quoted_table = quote_identifier(target_table)
|
||
column_sql = ", ".join(quote_identifier(name) for name, _ in pairs)
|
||
placeholders = ", ".join(["%s"] * len(pairs))
|
||
params = [value for _, value in pairs]
|
||
|
||
with connection.cursor() as cursor:
|
||
# 插入时只拼接列名，值全部走参数化。
|
||
cursor.execute(
|
||
f"INSERT INTO {quoted_table} ({column_sql}) VALUES ({placeholders})",
|
||
tuple(params),
|
||
)
|
||
affected_rows = int(cursor.rowcount or 0)
|
||
last_insert_id = cursor.lastrowid
|
||
connection.commit()
|
||
except Exception as exc: # noqa: BLE001
|
||
return browser_error_payload(str(exc))
|
||
|
||
clear_mysql_metrics_cache()
|
||
return {
|
||
"ok": True,
|
||
"updatedAt": utc_now(),
|
||
"message": f"已插入 {target_database}.{target_table}",
|
||
"affectedRows": affected_rows,
|
||
"lastInsertId": last_insert_id,
|
||
"error": "",
|
||
}
|
||
|
||
|
||
def update_mysql_row_payload(payload: dict[str, Any]) -> dict[str, Any]:
|
||
target_database = normalize_database_name(payload.get("database"))
|
||
target_table = normalize_table_name(payload.get("table"))
|
||
|
||
try:
|
||
with build_mysql_connection(target_database, autocommit=False) as connection:
|
||
columns = list_table_columns(connection, target_database, target_table)
|
||
pairs = normalize_write_values(payload.get("values"), columns)
|
||
if not pairs:
|
||
raise ValueError("at least one column value is required")
|
||
|
||
where_clause, where_params = build_identity_clause(payload.get("identity"), columns)
|
||
set_clause = ", ".join(f"{quote_identifier(name)} = %s" for name, _ in pairs)
|
||
params = [value for _, value in pairs] + where_params
|
||
|
||
with connection.cursor() as cursor:
|
||
# 更新固定只影响一行；identity 使用主键或唯一索引时语义最稳定。
|
||
cursor.execute(
|
||
f"UPDATE {quote_identifier(target_table)} SET {set_clause} WHERE {where_clause} LIMIT 1",
|
||
tuple(params),
|
||
)
|
||
affected_rows = int(cursor.rowcount or 0)
|
||
connection.commit()
|
||
except Exception as exc: # noqa: BLE001
|
||
return browser_error_payload(str(exc))
|
||
|
||
clear_mysql_metrics_cache()
|
||
return {
|
||
"ok": True,
|
||
"updatedAt": utc_now(),
|
||
"message": f"已更新 {target_database}.{target_table}",
|
||
"affectedRows": affected_rows,
|
||
"error": "",
|
||
}
|
||
|
||
|
||
def delete_mysql_row_payload(payload: dict[str, Any]) -> dict[str, Any]:
|
||
target_database = normalize_database_name(payload.get("database"))
|
||
target_table = normalize_table_name(payload.get("table"))
|
||
|
||
try:
|
||
with build_mysql_connection(target_database, autocommit=False) as connection:
|
||
columns = list_table_columns(connection, target_database, target_table)
|
||
where_clause, where_params = build_identity_clause(payload.get("identity"), columns)
|
||
|
||
with connection.cursor() as cursor:
|
||
# 删除同样只允许命中一行，避免前端误带了过宽的 where 条件。
|
||
cursor.execute(
|
||
f"DELETE FROM {quote_identifier(target_table)} WHERE {where_clause} LIMIT 1",
|
||
tuple(where_params),
|
||
)
|
||
affected_rows = int(cursor.rowcount or 0)
|
||
connection.commit()
|
||
except Exception as exc: # noqa: BLE001
|
||
return browser_error_payload(str(exc))
|
||
|
||
clear_mysql_metrics_cache()
|
||
return {
|
||
"ok": True,
|
||
"updatedAt": utc_now(),
|
||
"message": f"已删除 {target_database}.{target_table} 中的 1 行",
|
||
"affectedRows": affected_rows,
|
||
"error": "",
|
||
}
|
||
|
||
|
||
def add_mysql_column_payload(payload: dict[str, Any]) -> dict[str, Any]:
|
||
target_database = normalize_database_name(payload.get("database"))
|
||
target_table = normalize_table_name(payload.get("table"))
|
||
column_name = normalize_column_name(payload.get("columnName"))
|
||
column_type = normalize_sql_fragment(payload.get("columnType"), field_name="columnType")
|
||
nullable = bool(payload.get("nullable", True))
|
||
comment = str(payload.get("comment") or "").strip()
|
||
position = str(payload.get("position") or "last").strip().lower()
|
||
default_mode = str(payload.get("defaultMode") or "none").strip().lower()
|
||
default_value = payload.get("defaultValue")
|
||
|
||
clauses = [
|
||
f"ALTER TABLE {quote_identifier(target_table)} ADD COLUMN {quote_identifier(column_name)} {column_type}",
|
||
"NULL" if nullable else "NOT NULL",
|
||
]
|
||
params: list[Any] = []
|
||
|
||
# 默认值支持最常用的三种模式；更复杂表达式留给 SQL 工作台。
|
||
if default_mode == "literal":
|
||
clauses.append("DEFAULT %s")
|
||
params.append(default_value)
|
||
elif default_mode == "null":
|
||
clauses.append("DEFAULT NULL")
|
||
elif default_mode != "none":
|
||
raise ValueError(f"unsupported default mode: {default_mode}")
|
||
|
||
if comment:
|
||
clauses.append("COMMENT %s")
|
||
params.append(comment)
|
||
|
||
if position == "first":
|
||
clauses.append("FIRST")
|
||
elif position == "after":
|
||
after_column = normalize_column_name(payload.get("afterColumn"))
|
||
clauses.append(f"AFTER {quote_identifier(after_column)}")
|
||
elif position != "last":
|
||
raise ValueError(f"unsupported position: {position}")
|
||
|
||
try:
|
||
with build_mysql_connection(target_database, autocommit=False) as connection:
|
||
with connection.cursor() as cursor:
|
||
# 列类型本身是 SQL 语法片段，只做单语句校验，不做过度限制。
|
||
cursor.execute(" ".join(clauses), tuple(params))
|
||
connection.commit()
|
||
except Exception as exc: # noqa: BLE001
|
||
return browser_error_payload(str(exc))
|
||
|
||
clear_mysql_metrics_cache()
|
||
return {
|
||
"ok": True,
|
||
"updatedAt": utc_now(),
|
||
"message": f"已新增列 {target_table}.{column_name}",
|
||
"error": "",
|
||
}
|
||
|
||
|
||
def drop_mysql_column_payload(payload: dict[str, Any]) -> dict[str, Any]:
|
||
target_database = normalize_database_name(payload.get("database"))
|
||
target_table = normalize_table_name(payload.get("table"))
|
||
column_name = normalize_column_name(payload.get("columnName"))
|
||
|
||
try:
|
||
with build_mysql_connection(target_database, autocommit=False) as connection:
|
||
with connection.cursor() as cursor:
|
||
# 删除列是高影响操作，但用户明确要求支持，这里保持单列单次执行。
|
||
cursor.execute(
|
||
f"ALTER TABLE {quote_identifier(target_table)} DROP COLUMN {quote_identifier(column_name)}",
|
||
)
|
||
connection.commit()
|
||
except Exception as exc: # noqa: BLE001
|
||
return browser_error_payload(str(exc))
|
||
|
||
clear_mysql_metrics_cache()
|
||
return {
|
||
"ok": True,
|
||
"updatedAt": utc_now(),
|
||
"message": f"已删除列 {target_table}.{column_name}",
|
||
"error": "",
|
||
}
|
||
|
||
|
||
def add_mysql_index_payload(payload: dict[str, Any]) -> dict[str, Any]:
|
||
target_database = normalize_database_name(payload.get("database"))
|
||
target_table = normalize_table_name(payload.get("table"))
|
||
index_name = normalize_column_name(payload.get("indexName"))
|
||
index_mode = str(payload.get("indexMode") or "normal").strip().lower()
|
||
columns = payload.get("columns") if isinstance(payload.get("columns"), list) else []
|
||
normalized_columns = [normalize_column_name(item) for item in columns]
|
||
|
||
if not normalized_columns:
|
||
raise ValueError("at least one index column is required")
|
||
|
||
if index_mode == "unique":
|
||
prefix = f"ADD UNIQUE INDEX {quote_identifier(index_name)}"
|
||
elif index_mode == "fulltext":
|
||
prefix = f"ADD FULLTEXT INDEX {quote_identifier(index_name)}"
|
||
elif index_mode == "normal":
|
||
prefix = f"ADD INDEX {quote_identifier(index_name)}"
|
||
else:
|
||
raise ValueError(f"unsupported index mode: {index_mode}")
|
||
|
||
column_sql = ", ".join(quote_identifier(item) for item in normalized_columns)
|
||
|
||
try:
|
||
with build_mysql_connection(target_database, autocommit=False) as connection:
|
||
with connection.cursor() as cursor:
|
||
# 索引列名都走标识符转义，避免前端选择值被直接拼进 SQL。
|
||
cursor.execute(
|
||
f"ALTER TABLE {quote_identifier(target_table)} {prefix} ({column_sql})",
|
||
)
|
||
connection.commit()
|
||
except Exception as exc: # noqa: BLE001
|
||
return browser_error_payload(str(exc))
|
||
|
||
clear_mysql_metrics_cache()
|
||
return {
|
||
"ok": True,
|
||
"updatedAt": utc_now(),
|
||
"message": f"已新增索引 {target_table}.{index_name}",
|
||
"error": "",
|
||
}
|
||
|
||
|
||
def drop_mysql_index_payload(payload: dict[str, Any]) -> dict[str, Any]:
|
||
target_database = normalize_database_name(payload.get("database"))
|
||
target_table = normalize_table_name(payload.get("table"))
|
||
index_name = normalize_column_name(payload.get("indexName"))
|
||
|
||
try:
|
||
with build_mysql_connection(target_database, autocommit=False) as connection:
|
||
with connection.cursor() as cursor:
|
||
# PRIMARY KEY 的删除语法和普通索引不同，这里单独分支。
|
||
if index_name == "PRIMARY":
|
||
cursor.execute(
|
||
f"ALTER TABLE {quote_identifier(target_table)} DROP PRIMARY KEY",
|
||
)
|
||
else:
|
||
cursor.execute(
|
||
f"ALTER TABLE {quote_identifier(target_table)} DROP INDEX {quote_identifier(index_name)}",
|
||
)
|
||
connection.commit()
|
||
except Exception as exc: # noqa: BLE001
|
||
return browser_error_payload(str(exc))
|
||
|
||
clear_mysql_metrics_cache()
|
||
return {
|
||
"ok": True,
|
||
"updatedAt": utc_now(),
|
||
"message": f"已删除索引 {target_table}.{index_name}",
|
||
"error": "",
|
||
}
|
||
|
||
|
||
def execute_mysql_sql_payload(payload: dict[str, Any]) -> dict[str, Any]:
|
||
target_database = str(payload.get("database") or MYSQL_DEFAULT_DATABASE or "").strip() or None
|
||
sql = normalize_sql_text(payload.get("sql"))
|
||
kind = sql_statement_kind(sql)
|
||
mutating_kinds = {"insert", "update", "delete", "replace", "alter", "create", "drop", "truncate", "rename"}
|
||
|
||
try:
|
||
with build_mysql_connection(target_database, autocommit=False) as connection:
|
||
with connection.cursor() as cursor:
|
||
# SQL 工作台当前只执行单条语句，结果集最多保留固定行数。
|
||
cursor.execute(sql)
|
||
if cursor.description:
|
||
columns = [str(item[0] or "").strip() for item in cursor.description]
|
||
fetched_rows = cursor.fetchmany(MYSQL_QUERY_MAX_ROWS + 1)
|
||
truncated = len(fetched_rows) > MYSQL_QUERY_MAX_ROWS
|
||
rows = [normalize_mysql_row(item) for item in fetched_rows[:MYSQL_QUERY_MAX_ROWS]]
|
||
return {
|
||
"ok": True,
|
||
"updatedAt": utc_now(),
|
||
"database": target_database or "",
|
||
"kind": kind,
|
||
"columns": columns,
|
||
"rows": rows,
|
||
"rowCount": len(rows),
|
||
"truncated": truncated,
|
||
"error": "",
|
||
}
|
||
|
||
connection.commit()
|
||
affected_rows = int(cursor.rowcount or 0)
|
||
last_insert_id = cursor.lastrowid
|
||
except Exception as exc: # noqa: BLE001
|
||
return browser_error_payload(str(exc))
|
||
|
||
if kind in mutating_kinds:
|
||
clear_mysql_metrics_cache()
|
||
|
||
return {
|
||
"ok": True,
|
||
"updatedAt": utc_now(),
|
||
"database": target_database or "",
|
||
"kind": kind,
|
||
"affectedRows": affected_rows,
|
||
"lastInsertId": last_insert_id,
|
||
"message": f"{kind.upper()} 执行完成",
|
||
"error": "",
|
||
}
|