busya
525e0a56af
Root cause: fetch_available_datasources did not return database_id,
so frontend could not auto-set target_database_id when user selected
a datasource. User then had to manually pick a database on Target Config
tab — and could accidentally select the wrong one.
Backend changes:
- Add 'database_id' field to datasources response (from db_info.get('id'))
- Add _database_name tracking in SupersetSqlLabExecutor with getter
- Add database_name + database_backend to TargetSchemaValidationResponse
- Enrich resolve_database_id logging with database_name and all_keys
Frontend changes:
- In selectDatasource(), auto-set targetDatabaseId from ds.database_id
when present, so the correct target DB is used for schema checks.
353 lines
16 KiB
Python
353 lines
16 KiB
Python
# #region TargetSchemaValidation [C:4] [TYPE Module] [SEMANTICS translate, schema, validation, target-table]
|
||
# @BRIEF Проверка схемы целевой таблицы: запрос колонок через Superset SQL Lab,
|
||
# сравнение с ожидаемыми (из build_columns), возврат diff.
|
||
# @LAYER Service
|
||
# @RELATION DEPENDS_ON -> [SupersetSqlLabExecutor]
|
||
# @RELATION DEPENDS_ON -> [EXT:method:schemas.translate.TargetSchemaValidationRequest]
|
||
# @RELATION DEPENDS_ON -> [EXT:method:schemas.translate.TargetSchemaValidationResponse]
|
||
# @PRE Superset окружение доступно, target_database_id валиден.
|
||
# @POST Возвращает актуальные, ожидаемые, отсутствующие и лишние колонки.
|
||
# @SIDE_EFFECT Выполняет SQL-запрос через Superset SQL Lab.
|
||
# @RATIONALE C4 — оркестрация: вызов executor, парсинг ответа, diff и построение ответа.
|
||
# #endregion TargetSchemaValidation
|
||
|
||
import re
|
||
from typing import Any
|
||
|
||
from ...core.config_manager import ConfigManager
|
||
from ...core.logger import belief_scope, logger
|
||
from ...schemas.translate import (
|
||
TargetSchemaColumnInfo,
|
||
TargetSchemaValidationRequest,
|
||
TargetSchemaValidationResponse,
|
||
)
|
||
from .superset_executor import SupersetSqlLabExecutor
|
||
|
||
_TABLE_NAME_RE = re.compile(r'^[a-zA-Z_][a-zA-Z0-9_]*$')
|
||
|
||
|
||
# #region _build_expected_columns [C:2] [TYPE Function]
|
||
# @BRIEF Собирает список ожидаемых колонок по конфигурации column mapping.
|
||
def _build_expected_columns(req: TargetSchemaValidationRequest) -> list[TargetSchemaColumnInfo]:
|
||
"""Определяет, какие колонки INSERT будет пытаться заполнить."""
|
||
names: list[str] = []
|
||
|
||
if req.target_key_cols:
|
||
names.extend(req.target_key_cols)
|
||
|
||
effective_target = req.target_column or req.translation_column
|
||
if effective_target:
|
||
names.append(effective_target)
|
||
|
||
if req.target_language_column:
|
||
names.append(req.target_language_column)
|
||
|
||
if req.target_source_column:
|
||
names.append(req.target_source_column)
|
||
|
||
if req.target_source_language_column:
|
||
names.append(req.target_source_language_column)
|
||
|
||
names.append("context")
|
||
names.append("is_original")
|
||
|
||
seen: set[str] = set()
|
||
deduped: list[str] = []
|
||
for c in names:
|
||
if c and c not in seen:
|
||
deduped.append(c)
|
||
seen.add(c)
|
||
|
||
return [TargetSchemaColumnInfo(name=n) for n in deduped]
|
||
# #endregion _build_expected_columns
|
||
|
||
|
||
# #region _extract_columns_from_rows [C:2] [TYPE Function]
|
||
# @BRIEF Извлекает список колонок таблицы из data-строк результата SQL Lab.
|
||
#
|
||
# Для PostgreSQL (information_schema.columns) каждая строка содержит:
|
||
# {"column_name": "id", "data_type": "integer", "is_nullable": "YES"}
|
||
#
|
||
# Для ClickHouse (DESCRIBE TABLE) каждая строка содержит:
|
||
# {"name": "id", "type": "Int32"}
|
||
#
|
||
# Функция ищет имя колонки по нескольким возможным ключам и возвращает
|
||
# унифицированный список TargetSchemaColumnInfo.
|
||
def _extract_columns_from_rows(data_rows: list[dict[str, Any]]) -> list[dict[str, Any]]:
|
||
"""
|
||
Извлекает информацию о колонках таблицы из data-строк SQL Lab ответа.
|
||
"""
|
||
columns_info: list[dict[str, Any]] = []
|
||
|
||
for row in data_rows:
|
||
if not isinstance(row, dict):
|
||
continue
|
||
|
||
# Имя колонки: ищем по всем возможным ключам (PG, CH, разные версии Superset)
|
||
col_name = (
|
||
row.get("column_name") # PostgreSQL information_schema
|
||
or row.get("name") # ClickHouse DESCRIBE TABLE / MySQL SHOW COLUMNS
|
||
or row.get("Field") # MySQL SHOW COLUMNS
|
||
)
|
||
if not col_name:
|
||
continue
|
||
|
||
# Тип данных
|
||
col_type = (
|
||
row.get("data_type") # PG information_schema
|
||
or row.get("type") # ClickHouse DESCRIBE TABLE
|
||
or row.get("Type") # MySQL SHOW COLUMNS
|
||
)
|
||
|
||
# Nullable
|
||
is_nullable = row.get("is_nullable") # PG
|
||
if isinstance(is_nullable, str):
|
||
is_nullable = is_nullable.upper() == "YES"
|
||
elif not isinstance(is_nullable, bool):
|
||
is_nullable = True # ClickHouse не возвращает is_nullable, считаем True
|
||
|
||
columns_info.append({
|
||
"name": str(col_name),
|
||
"type": str(col_type) if col_type else None,
|
||
"is_nullable": is_nullable,
|
||
})
|
||
|
||
return columns_info
|
||
# #endregion _extract_columns_from_rows
|
||
|
||
|
||
# #region _parse_sqllab_result [C:2] [TYPE Function]
|
||
# @BRIEF Извлекает data-строки из ответа Superset SQL Lab.
|
||
# Поддерживает 3 формата: sync mode, async polling, get_query_results.
|
||
def _parse_sqllab_result(result: dict[str, Any]) -> tuple[list[dict[str, Any]], bool]:
|
||
"""
|
||
Извлекает строки данных из ответа SQL Lab.
|
||
Returns (data_rows, table_exists).
|
||
"""
|
||
if not result:
|
||
return [], False
|
||
|
||
raw = result.get("raw_response", result)
|
||
if not raw:
|
||
return [], False
|
||
|
||
data_rows = None
|
||
|
||
# Формат 1: прямой ответ с data
|
||
if "data" in raw and isinstance(raw["data"], list):
|
||
data_rows = raw["data"]
|
||
|
||
# Формат 2: вложенный result (async polling)
|
||
elif "result" in raw and isinstance(raw["result"], dict):
|
||
res = raw["result"]
|
||
data_rows = res.get("data") if isinstance(res.get("data"), list) else None
|
||
|
||
# Формат 3: ключ results (get_query_results)
|
||
elif "results" in raw and isinstance(raw["results"], dict):
|
||
res = raw["results"]
|
||
if "data" in res and isinstance(res["data"], list):
|
||
data_rows = res["data"]
|
||
elif isinstance(res.get("result"), dict):
|
||
sub = res["result"]
|
||
data_rows = sub.get("data") if isinstance(sub.get("data"), list) else None
|
||
|
||
# Формат 4: results — список строк (execute_and_poll sync mode возвращает так)
|
||
elif "results" in raw and isinstance(raw["results"], list):
|
||
data_rows = raw["results"]
|
||
|
||
if data_rows is None:
|
||
return [], False
|
||
|
||
table_exists = len(data_rows) > 0
|
||
|
||
# Нормализуем формат строк: если data_rows — список списков, а не словарей
|
||
# (бывает в некоторых версиях Superset), конвертируем через columns metadata
|
||
if data_rows and not isinstance(data_rows[0], dict):
|
||
columns_raw = raw.get("columns") or (raw.get("result") or {}).get("columns") or []
|
||
col_names = []
|
||
for c in columns_raw:
|
||
cn = c.get("name") or c.get("column_name") or ""
|
||
if cn:
|
||
col_names.append(cn)
|
||
if col_names:
|
||
normalized = []
|
||
for row in data_rows:
|
||
if isinstance(row, list) and len(row) == len(col_names):
|
||
normalized.append(dict(zip(col_names, row)))
|
||
data_rows = normalized
|
||
|
||
return data_rows, table_exists
|
||
# #endregion _parse_sqllab_result
|
||
|
||
|
||
# #region validate_target_table_schema [C:4] [TYPE Function] [SEMANTICS translate, schema, validate, orchestrate]
|
||
# @BRIEF Основная функция: проверяет схему целевой таблицы через Superset SQL Lab.
|
||
# @PRE Superset окружение и target_database_id валидны.
|
||
# @POST Возвращает TargetSchemaValidationResponse с diff-анализом.
|
||
# @SIDE_EFFECT Выполняет SQL-запрос к information_schema через Superset SQL Lab API.
|
||
# @RELATION DEPENDS_ON -> [_build_expected_columns]
|
||
# @RELATION DEPENDS_ON -> [_extract_columns_from_rows]
|
||
# @RELATION DEPENDS_ON -> [_parse_sqllab_result]
|
||
# @RELATION DEPENDS_ON -> [SupersetSqlLabExecutor]
|
||
def validate_target_table_schema(
|
||
req: TargetSchemaValidationRequest,
|
||
config_manager: ConfigManager,
|
||
) -> TargetSchemaValidationResponse:
|
||
"""
|
||
Проверяет схему целевой таблицы:
|
||
1. Вычисляет ожидаемые колонки
|
||
2. Запрашивает актуальные через Superset SQL Lab
|
||
3. Сравнивает и возвращает diff
|
||
"""
|
||
with belief_scope("validate_target_table_schema"):
|
||
# 1. Ожидаемые колонки
|
||
expected = _build_expected_columns(req)
|
||
expected_names = {c.name for c in expected}
|
||
logger.reason("Built expected columns",
|
||
extra={"payload": {"count": len(expected), "columns": [c.name for c in expected]}})
|
||
|
||
# 2. Валидация имён (SQL injection protection)
|
||
if req.target_table and not _TABLE_NAME_RE.match(req.target_table):
|
||
logger.explore("Invalid target table name rejected",
|
||
extra={"payload": {"table": req.target_table},
|
||
"error": "Invalid characters in table name"})
|
||
return TargetSchemaValidationResponse(
|
||
table_exists=False,
|
||
error=f"Invalid target table name: '{req.target_table}'. Only alphanumeric and underscore allowed.",
|
||
expected_columns=expected, actual_columns=[],
|
||
missing_columns=expected, extra_columns=[], all_present=False,
|
||
)
|
||
if req.target_schema and not _TABLE_NAME_RE.match(req.target_schema):
|
||
logger.explore("Invalid target schema name rejected",
|
||
extra={"payload": {"schema": req.target_schema},
|
||
"error": "Invalid characters in schema name"})
|
||
return TargetSchemaValidationResponse(
|
||
table_exists=False,
|
||
error=f"Invalid target schema name: '{req.target_schema}'. Only alphanumeric and underscore allowed.",
|
||
expected_columns=expected, actual_columns=[],
|
||
missing_columns=expected, extra_columns=[], all_present=False,
|
||
)
|
||
|
||
# 3. Запрос через Superset SQL Lab
|
||
target_ref = f"{req.target_schema or 'public'}.{req.target_table}"
|
||
logger.reason("Querying target table schema",
|
||
extra={"payload": {"table": target_ref, "env": req.environment_id, "db_id": req.target_database_id}})
|
||
|
||
try:
|
||
executor = SupersetSqlLabExecutor(config_manager, req.environment_id)
|
||
executor.resolve_database_id(target_database_id=req.target_database_id)
|
||
backend = executor.get_database_backend() or ""
|
||
db_name = executor.get_database_name() or "unknown"
|
||
|
||
safe_schema = (req.target_schema or "public").replace("'", "''")
|
||
safe_table = req.target_table.replace("'", "''")
|
||
|
||
# Нормализуем backend через маппинг (как в get_dialect_from_database)
|
||
# чтобы правильно обработать "clickhousedb" → "clickhouse",
|
||
# "greenplum" → "postgresql" и другие варианты
|
||
_backend_normalize = {
|
||
"clickhouse": "clickhouse",
|
||
"clickhousedb": "clickhouse",
|
||
"postgresql": "postgresql",
|
||
"greenplum": "postgresql",
|
||
"mysql": "mysql",
|
||
"mssql": "mssql",
|
||
"sqlite": "sqlite",
|
||
"oracle": "oracle",
|
||
"snowflake": "snowflake",
|
||
"bigquery": "bigquery",
|
||
"redshift": "redshift",
|
||
"presto": "presto",
|
||
"trino": "trino",
|
||
"druid": "druid",
|
||
"hive": "hive",
|
||
"spark": "spark",
|
||
"databricks": "databricks",
|
||
}
|
||
normalized_backend = _backend_normalize.get(backend.lower().strip(), backend.lower().strip())
|
||
# ClickHouse использует system.columns, всё остальное — information_schema.columns
|
||
is_clickhouse = normalized_backend == "clickhouse"
|
||
if is_clickhouse:
|
||
sql = (
|
||
f"SELECT name, type, default_expression AS data_default "
|
||
f"FROM system.columns "
|
||
f"WHERE database = '{safe_schema}' "
|
||
f"AND table = '{safe_table}' "
|
||
f"ORDER BY position"
|
||
)
|
||
else:
|
||
sql = (
|
||
f"SELECT column_name, data_type, is_nullable "
|
||
f"FROM information_schema.columns "
|
||
f"WHERE table_schema = '{safe_schema}' "
|
||
f"AND table_name = '{safe_table}' "
|
||
f"ORDER BY ordinal_position"
|
||
)
|
||
|
||
logger.reason("Executing SQL Lab query",
|
||
extra={"payload": {"sql": sql[:200], "backend": backend, "database_name": db_name}})
|
||
|
||
result = executor.execute_and_poll(sql=sql, max_polls=15, poll_interval_seconds=1.0)
|
||
status = result.get("status", "")
|
||
|
||
if status == "failed":
|
||
error_msg = result.get("error_message", "SQL Lab query failed")
|
||
logger.explore("SQL Lab query failed",
|
||
extra={"payload": {"table": target_ref}, "error": error_msg})
|
||
return TargetSchemaValidationResponse(
|
||
table_exists=False, error=f"Superset SQL Lab error: {error_msg}",
|
||
expected_columns=expected, actual_columns=[],
|
||
missing_columns=expected, extra_columns=[], all_present=False,
|
||
database_name=db_name, database_backend=backend,
|
||
)
|
||
|
||
# Парсим data-строки результата (реальные колонки таблицы)
|
||
data_rows, table_exists = _parse_sqllab_result(result)
|
||
actual_cols_raw = _extract_columns_from_rows(data_rows)
|
||
|
||
actual_cols = []
|
||
for col in actual_cols_raw:
|
||
actual_cols.append(TargetSchemaColumnInfo(
|
||
name=col["name"],
|
||
data_type=col.get("type"),
|
||
is_nullable=col.get("is_nullable", True),
|
||
))
|
||
|
||
actual_names = {c.name for c in actual_cols}
|
||
|
||
# 4. Diff
|
||
missing = [col for col in expected if col.name not in actual_names]
|
||
extra = [col for col in actual_cols if col.name not in expected_names]
|
||
|
||
logger.reflect("Schema check complete",
|
||
extra={"payload": {
|
||
"table": target_ref, "table_exists": table_exists,
|
||
"expected": len(expected), "found": len(actual_cols),
|
||
"missing": len(missing), "extra": len(extra),
|
||
"all_present": len(missing) == 0,
|
||
"actual_columns": [c.name for c in actual_cols],
|
||
}})
|
||
|
||
return TargetSchemaValidationResponse(
|
||
table_exists=table_exists,
|
||
expected_columns=expected,
|
||
actual_columns=actual_cols,
|
||
missing_columns=missing,
|
||
extra_columns=extra,
|
||
all_present=len(missing) == 0,
|
||
database_name=db_name, database_backend=backend,
|
||
)
|
||
|
||
except Exception as e:
|
||
logger.explore("Target schema validation failed",
|
||
extra={"payload": {"table": target_ref, "env": req.environment_id},
|
||
"error": str(e)})
|
||
return TargetSchemaValidationResponse(
|
||
table_exists=False, error=f"Failed to validate target schema: {e}",
|
||
expected_columns=expected, actual_columns=[],
|
||
missing_columns=expected, extra_columns=[], all_present=False,
|
||
database_name=db_name, database_backend=backend,
|
||
)
|
||
# #endregion validate_target_table_schema
|
||
# #endregion TargetSchemaValidation
|