busya/ss-tools
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(llm): add fetch-models endpoint, fix SQL Lab INSERT (client_id truncation, sync mode, target_column, timestamp normalization)

Browse Source 
- Add POST /api/llm/providers/fetch-models route with LLMClient.fetch_models()
- Add target_column to TranslationJob model/schema/service/orchestrator
- Fix SQL Lab execute: truncate client_id to 11 chars (varchar(11))
- Switch SQL Lab to sync mode (runAsync: false) — no Celery workers
- Fix polling: unwrap nested result from Superset query API
- Fix ClickHouse timestamp: normalize float timestamps to YYYY-MM-DD
This commit is contained in:
busya
2026-05-13 20:06:15 +03:00
parent 39ab647851
commit a3c7c402b7
276 changed files with 1923 additions and 1822 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
backend/src/plugins/git/llm_extension.py
View File

@@ -1,5 +1,4 @@
# #region GitLLMExtensionModule [C:3] [TYPE Module] [SEMANTICS git, llm, commit, message, generation]
# @COMPLEXITY: 3
# @BRIEF LLM-based extensions for the Git plugin, specifically for commit message generation.
# @LAYER: Domain
# @RELATION DEPENDS_ON -> [LLMClient]

1
backend/src/plugins/llm_analysis/__tests__/test_client_headers.py
View File

@@ -1,6 +1,5 @@
# region TestClientHeaders [TYPE Module]
# @RELATION: BELONGS_TO -> SrcRoot
# @COMPLEXITY: 3
# @SEMANTICS: tests, llm-client, openrouter, headers
# @PURPOSE: Verify OpenRouter client initialization includes provider-specific headers.

1
backend/src/plugins/llm_analysis/__tests__/test_screenshot_service.py
View File

@@ -1,6 +1,5 @@
# region TestScreenshotService [TYPE Module]
# @RELATION: VERIFIES ->[src.plugins.llm_analysis.service.ScreenshotService]
# @COMPLEXITY: 3
# @SEMANTICS: tests, screenshot-service, navigation, timeout-regression
# @PURPOSE: Protect dashboard screenshot navigation from brittle networkidle waits.

1
backend/src/plugins/llm_analysis/__tests__/test_service.py
View File

@@ -1,6 +1,5 @@
# region TestService [TYPE Module]
# @RELATION: BELONGS_TO -> SrcRoot
# @COMPLEXITY: 3
# @SEMANTICS: tests, llm-analysis, fallback, provider-error, unknown-status
# @PURPOSE: Verify LLM analysis transport/provider failures do not masquerade as dashboard FAIL results.

1
backend/src/plugins/llm_analysis/plugin.py
View File

@@ -1,5 +1,4 @@
# #region LLMAnalysisPlugin [C:3] [TYPE Module] [SEMANTICS llm, analysis, dashboard, validation, documentation]
# @COMPLEXITY: 3
# @BRIEF Implements DashboardValidationPlugin and DocumentationPlugin.
# @LAYER: Domain
# @RELATION INHERITS -> [PluginBase]

1
backend/src/plugins/llm_analysis/scheduler.py
View File

@@ -1,5 +1,4 @@
# #region LLMAnalysisScheduler [C:3] [TYPE Module] [SEMANTICS scheduler, llm, validation, task, cron]
# @COMPLEXITY: 3
# @BRIEF Provides helper functions to schedule LLM-based validation tasks.
# @LAYER: Domain
# @RELATION DEPENDS_ON -> [SchedulerService]

24
backend/src/plugins/llm_analysis/service.py
View File

@@ -1,5 +1,4 @@
# #region LLMAnalysisService [C:3] [TYPE Module] [SEMANTICS llm, screenshot, playwright, openai, tenacity]
# @COMPLEXITY: 3
# @BRIEF Services for LLM interaction and dashboard screenshots.
# @LAYER: Domain
# @RELATION DEPENDS_ON -> playwright
@@ -876,6 +875,29 @@ class LLMClient:
return await self.get_json_completion(messages)
# endregion LLMClient.test_runtime_connection
# region LLMClient.fetch_models [TYPE Function]
# @PURPOSE: Fetch available models from the provider's API.
# @PRE: Client is initialized with provider credentials.
# @POST: Returns a list of model ID strings.
# @SIDE_EFFECT: Calls external LLM API /v1/models endpoint.
async def fetch_models(self) -> List[str]:
with belief_scope("LLMClient.fetch_models"):
try:
response = await self.client.models.list()
model_ids = [m.id for m in response.data]
model_ids.sort()
logger.reason(
f"[LLMClient.fetch_models] Fetched {len(model_ids)} models from {self.base_url}",
extra={"src": "LLMClient.fetch_models"},
)
return model_ids
except Exception as e:
logger.warning(
f"[LLMClient.fetch_models] Failed to fetch models: {e}",
)
raise
# endregion LLMClient.fetch_models
# region LLMClient.analyze_dashboard [TYPE Function]
# @PURPOSE: Sends dashboard data (screenshot + logs) to LLM for health analysis.
# @PRE: screenshot_path exists, logs is a list of strings.

3
backend/src/plugins/translate/__tests__/test_clickhouse_insert_integration.py
View File

@@ -1,5 +1,4 @@
# region ClickHouseInsertIntegration [TYPE Module]
# @COMPLEXITY: 3
# @SEMANTICS: test, clickhouse, integration, insert, join
# @PURPOSE: Integration tests for ClickHouse INSERT with timestamp normalization and JOIN verification.
# @LAYER: Test
@@ -369,6 +368,8 @@ class TestOrchestratorInsertFlow:
) as MockExecutor:
mock_executor = MagicMock()
MockExecutor.return_value = mock_executor
mock_executor.resolve_database_id.return_value = None
mock_executor.get_database_backend.return_value = None
mock_executor.execute_and_poll.return_value = {
"status": "success",
"query_id": "q-123",

2
backend/src/plugins/translate/__tests__/test_dictionary.py
View File

@@ -1,5 +1,4 @@
# region DictionaryTests [TYPE Module]
# @COMPLEXITY: 3
# @SEMANTICS: tests, dictionary, crud, import, filter
# @PURPOSE: Validate DictionaryManager CRUD, import, deletion guards, and batch filtering.
# @RELATION: BINDS_TO -> [DictionaryManager:Class]
@@ -38,7 +37,6 @@ from src.plugins.translate._utils import _normalize_term, _detect_delimiter
# region _FakeJob [TYPE Class]
# @COMPLEXITY: 1
# @PURPOSE: Helper to create inline TranslationJob records.
class _FakeJob:
pass

208
backend/src/plugins/translate/__tests__/test_orchestrator.py
View File

@@ -1,5 +1,4 @@
# region OrchestratorTests [TYPE Module]
# @COMPLEXITY: 3
# @SEMANTICS: test, translate, orchestrator, events
# @PURPOSE: Tests for TranslationOrchestrator: run lifecycle, partial failure, batch retry, event invariants, NULL handling.
# @LAYER: Test
@@ -10,9 +9,10 @@
# @TEST_FIXTURE: mock_config_manager -> MagicMock ConfigManager
# @TEST_EDGE: missing_preview -> raises ValueError
# @TEST_EDGE: invalid_run_status -> raises ValueError
# @TEST_EDGE: executor_failure -> run is marked FAILED
import pytest
from unittest.mock import MagicMock
from unittest.mock import MagicMock, patch
from datetime import datetime, timezone
from src.plugins.translate.orchestrator import TranslationOrchestrator
@@ -234,6 +234,210 @@ class TestTranslationOrchestrator:
with pytest.raises(ValueError, match="PENDING"):
orch.execute_run(run)
# region test_execute_run_success [TYPE Function]
# @BRIEF Happy path: executor completes, SQL generated, Superset submits.
def test_execute_run_success(self, mock_job: MagicMock) -> None:
db = MagicMock()
config_manager = MagicMock()
mock_job.target_table = "target_tbl"
# Mock job query — first call returns mock_job, final re-query returns completed_run
completed_run_response = MagicMock()
completed_run_response.id = "run-1"
completed_run_response.job_id = "job-123"
completed_run_response.status = "COMPLETED"
completed_run_response.total_records = 10
completed_run_response.successful_records = 10
completed_run_response.failed_records = 0
completed_run_response.skipped_records = 0
completed_run_response.completed_at = datetime.now(timezone.utc)
completed_run_response.error_message = None
completed_run_response.insert_status = "success"
completed_run_response.superset_execution_id = "q-1"
completed_run_response.superset_execution_log = {}
# query job -> mock_job, query TranslationRun -> completed_run_response
db.query.return_value.filter.return_value.first.side_effect = [
mock_job,
completed_run_response,
]
# Create a PENDING input run
run = MagicMock()
run.id = "run-1"
run.job_id = "job-123"
run.status = "PENDING"
with patch(
"src.plugins.translate.orchestrator.TranslationExecutor"
) as MockExecutor:
mock_executor_instance = MagicMock()
mock_executor_instance.execute_run.return_value = completed_run_response
MockExecutor.return_value = mock_executor_instance
orch = TranslationOrchestrator(db, config_manager, "test-user")
with patch.object(orch, "event_log"):
with patch.object(
orch, "_generate_and_insert_sql",
return_value={"status": "success", "query_id": "q-1", "rows_affected": 10},
):
result = orch.execute_run(run)
assert result.status == "COMPLETED"
assert result.insert_status == "success"
assert result.superset_execution_id == "q-1"
assert result.total_records == 10
assert result.successful_records == 10
# endregion test_execute_run_success
# region test_execute_run_executor_failure [TYPE Function]
# @BRIEF Executor raises exception, run is marked FAILED.
def test_execute_run_executor_failure(self, mock_job: MagicMock) -> None:
db = MagicMock()
config_manager = MagicMock()
# Mock job query
db.query.return_value.filter.return_value.first.return_value = mock_job
run = MagicMock()
run.id = "run-1"
run.job_id = "job-123"
run.status = "PENDING"
run.error_message = None
with patch(
"src.plugins.translate.orchestrator.TranslationExecutor"
) as MockExecutor:
mock_executor_instance = MagicMock()
mock_executor_instance.execute_run.side_effect = ValueError(
"LLM provider unavailable"
)
MockExecutor.return_value = mock_executor_instance
orch = TranslationOrchestrator(db, config_manager, "test-user")
with patch.object(orch, "event_log"):
result = orch.execute_run(run)
assert result.status == "FAILED"
assert "LLM provider unavailable" in (result.error_message or "")
# endregion test_execute_run_executor_failure
# region test_execute_run_skip_insert [TYPE Function]
# @BRIEF skip_insert=True completes run without SQL generation or Superset submission.
def test_execute_run_skip_insert(self, mock_job: MagicMock) -> None:
db = MagicMock()
config_manager = MagicMock()
db.query.return_value.filter.return_value.first.return_value = mock_job
run = MagicMock()
run.id = "run-1"
run.job_id = "job-123"
run.status = "PENDING"
run.completed_at = None
completed_run = MagicMock()
completed_run.id = "run-1"
completed_run.job_id = "job-123"
completed_run.status = "COMPLETED"
completed_run.total_records = 5
completed_run.successful_records = 5
completed_run.failed_records = 0
completed_run.skipped_records = 0
completed_run.completed_at = datetime.now(timezone.utc)
completed_run.error_message = None
with patch(
"src.plugins.translate.orchestrator.TranslationExecutor"
) as MockExecutor:
mock_executor_instance = MagicMock()
mock_executor_instance.execute_run.return_value = completed_run
MockExecutor.return_value = mock_executor_instance
orch = TranslationOrchestrator(db, config_manager, "test-user")
with patch.object(orch, "_generate_and_insert_sql") as mock_gen_sql:
with patch.object(orch, "event_log"):
result = orch.execute_run(run, skip_insert=True)
assert result.status == "COMPLETED"
# _generate_and_insert_sql should NOT be called in skip_insert mode
mock_gen_sql.assert_not_called()
assert result.total_records == 5
# endregion test_execute_run_skip_insert
# region test_execute_run_no_job [TYPE Function]
# @BRIEF Job not found raises ValueError.
def test_execute_run_no_job(self) -> None:
db = MagicMock()
config_manager = MagicMock()
# Mock job query returns None (job not found)
db.query.return_value.filter.return_value.first.return_value = None
run = MagicMock()
run.id = "run-1"
run.job_id = "job-missing"
run.status = "PENDING"
orch = TranslationOrchestrator(db, config_manager, "test-user")
with pytest.raises(ValueError, match="not found"):
orch.execute_run(run)
# endregion test_execute_run_no_job
# region test_execute_run_insert_failure [TYPE Function]
# @BRIEF SQL generation/insert returns failure, run still completes but with error logged.
def test_execute_run_insert_failure(self, mock_job: MagicMock) -> None:
db = MagicMock()
config_manager = MagicMock()
# First query returns mock_job, re-query returns completed_run
completed_run_response = MagicMock()
completed_run_response.id = "run-1"
completed_run_response.job_id = "job-123"
completed_run_response.status = "COMPLETED"
completed_run_response.total_records = 3
completed_run_response.successful_records = 3
completed_run_response.failed_records = 0
completed_run_response.skipped_records = 0
completed_run_response.completed_at = datetime.now(timezone.utc)
completed_run_response.error_message = None
completed_run_response.insert_status = "failed"
completed_run_response.superset_execution_id = ""
completed_run_response.superset_execution_log = {}
db.query.return_value.filter.return_value.first.side_effect = [
mock_job,
completed_run_response,
]
run = MagicMock()
run.id = "run-1"
run.job_id = "job-123"
run.status = "PENDING"
with patch(
"src.plugins.translate.orchestrator.TranslationExecutor"
) as MockExecutor:
mock_executor_instance = MagicMock()
mock_executor_instance.execute_run.return_value = completed_run_response
MockExecutor.return_value = mock_executor_instance
orch = TranslationOrchestrator(db, config_manager, "test-user")
with patch.object(
orch, "_generate_and_insert_sql",
return_value={"status": "failed", "error_message": "timeout", "query_id": None},
):
with patch.object(orch, "event_log"):
result = orch.execute_run(run)
assert result.status == "COMPLETED"
assert result.insert_status == "failed"
assert "timeout" in (result.error_message or "")
# endregion test_execute_run_insert_failure
# region test_cancel_run [TYPE Function]
# @PURPOSE: Cancel a run changes status to CANCELLED.
def test_cancel_run(self) -> None:

1
backend/src/plugins/translate/__tests__/test_preview.py
View File

@@ -1,5 +1,4 @@
# region TranslationPreviewTests [TYPE Module]
# @COMPLEXITY: 3
# @SEMANTICS: test, translate, preview, session
# @PURPOSE: Test TranslationPreview service: preview, approve/edit/reject state transitions, cost estimation, acceptance gate.
# @LAYER: Test

1
backend/src/plugins/translate/__tests__/test_sql_generator.py
View File

@@ -1,5 +1,4 @@
# region SQLGeneratorTests [TYPE Module]
# @COMPLEXITY: 3
# @SEMANTICS: test, translate, sql_generator
# @PURPOSE: Tests for SQLGenerator: PostgreSQL INSERT + UPSERT, ClickHouse INSERT, dialect quoting, NULL handling, injection safety.
# @LAYER: Test

14
backend/src/plugins/translate/dictionary.py
View File

@@ -32,7 +32,6 @@ from ._utils import _normalize_term, _detect_delimiter
# @SIDE_EFFECT: Creates, updates, deletes TerminologyDictionary and DictionaryEntry rows; enforces deletion guards.
class DictionaryManager:
# region DictionaryManager.create_dictionary [TYPE Function]
# @COMPLEXITY: 3
# @PURPOSE: Create a new terminology dictionary.
# @PRE: payload contains name, source_dialect, target_dialect.
# @POST: New TerminologyDictionary row is created and returned.
@@ -60,7 +59,6 @@ class DictionaryManager:
# endregion DictionaryManager.create_dictionary
# region DictionaryManager.update_dictionary [TYPE Function]
# @COMPLEXITY: 3
# @PURPOSE: Update an existing terminology dictionary.
# @PRE: dict_id exists in terminology_dictionaries table.
# @POST: Dictionary metadata is updated and returned.
@@ -92,7 +90,6 @@ class DictionaryManager:
# endregion DictionaryManager.update_dictionary
# region DictionaryManager.delete_dictionary [TYPE Function]
# @COMPLEXITY: 4
# @PURPOSE: Delete a dictionary, blocked if attached to active/scheduled jobs.
# @PRE: dict_id exists.
# @POST: Dictionary and its entries are deleted, unless attached to ACTIVE/READY/RUNNING/SCHEDULED jobs.
@@ -136,7 +133,6 @@ class DictionaryManager:
# endregion DictionaryManager.delete_dictionary
# region DictionaryManager.get_dictionary [TYPE Function]
# @COMPLEXITY: 2
# @PURPOSE: Get a single dictionary by ID with entry count.
# @PRE: dict_id exists.
# @POST: Returns dict with dictionary + entry_count or raises ValueError.
@@ -149,7 +145,6 @@ class DictionaryManager:
# endregion DictionaryManager.get_dictionary
# region DictionaryManager.list_dictionaries [TYPE Function]
# @COMPLEXITY: 2
# @PURPOSE: List dictionaries with pagination and entry counts.
# @PRE: page >= 1, page_size between 1 and 100.
# @POST: Returns (list of dicts, total_count).
@@ -169,7 +164,6 @@ class DictionaryManager:
# endregion DictionaryManager.list_dictionaries
# region DictionaryManager.add_entry [TYPE Function]
# @COMPLEXITY: 3
# @PURPOSE: Add an entry to a dictionary, enforcing unique source_term_normalized.
# @PRE: dict_id exists. source_term and target_term are non-empty.
# @POST: New DictionaryEntry row is created or raises on duplicate.
@@ -210,7 +204,6 @@ class DictionaryManager:
# endregion DictionaryManager.add_entry
# region DictionaryManager.edit_entry [TYPE Function]
# @COMPLEXITY: 3
# @PURPOSE: Edit an existing dictionary entry with duplicate-aware normalization.
# @PRE: entry_id exists.
# @POST: Entry fields are updated.
@@ -255,7 +248,6 @@ class DictionaryManager:
# endregion DictionaryManager.edit_entry
# region DictionaryManager.delete_entry [TYPE Function]
# @COMPLEXITY: 2
# @PURPOSE: Delete a single dictionary entry.
# @PRE: entry_id exists.
# @POST: Entry is deleted.
@@ -272,7 +264,6 @@ class DictionaryManager:
# endregion DictionaryManager.delete_entry
# region DictionaryManager.clear_entries [TYPE Function]
# @COMPLEXITY: 2
# @PURPOSE: Delete all entries for a dictionary.
# @PRE: dict_id exists.
# @POST: All entries for the dictionary are deleted.
@@ -290,7 +281,6 @@ class DictionaryManager:
# endregion DictionaryManager.clear_entries
# region DictionaryManager.list_entries [TYPE Function]
# @COMPLEXITY: 2
# @PURPOSE: List entries for a dictionary with pagination.
# @PRE: dict_id exists.
# @POST: Returns (list of entries, total_count).
@@ -316,7 +306,6 @@ class DictionaryManager:
# endregion DictionaryManager.list_entries
# region DictionaryManager.import_entries [TYPE Function]
# @COMPLEXITY: 4
# @PURPOSE: Import entries from CSV/TSV content with duplicate detection and conflict resolution.
# @PRE: content is valid CSV or TSV. dict_id exists.
# @POST: Entries are created/updated/skipped per conflict mode. Returns result summary.
@@ -456,7 +445,6 @@ class DictionaryManager:
# endregion DictionaryManager.import_entries
# region DictionaryManager.filter_for_batch [TYPE Function]
# @COMPLEXITY: 4
# @PURPOSE: Scan batch texts for case-insensitive, word-boundary-aware matches against all dictionaries attached to a job.
# @PRE: job_id exists and source_texts is a list of strings.
# @POST: Returns list of matched entries with match info, sorted by priority across attached dictionaries.
@@ -555,7 +543,6 @@ class DictionaryManager:
# region DictionaryManager.submit_correction [TYPE Function]
# @COMPLEXITY: 4
# @PURPOSE: Submit a term correction from a run result. Validates language match, detects conflicts.
# @PRE: source_term, incorrect_target_term, corrected_target_term are non-empty. dict_id exists.
# @POST: Entry created or updated; origin tracking populated. Returns action + conflict info.
@@ -653,7 +640,6 @@ class DictionaryManager:
# endregion DictionaryManager.submit_correction
# region DictionaryManager.submit_bulk_corrections [TYPE Function]
# @COMPLEXITY: 4
# @PURPOSE: Submit multiple term corrections atomically — all succeed or all fail.
# @PRE: corrections list is non-empty. dict_id exists.
# @POST: All corrections applied or none applied with conflict list.

1
backend/src/plugins/translate/events.py
View File

@@ -48,7 +48,6 @@ class TranslationEventLog:
self.db = db
# region log_event [TYPE Function]
# @COMPLEXITY: 4
# @PURPOSE: Write an immutable event. Enforces terminal event invariant for non-null run_id.
# @PRE: event_type must be a known type. If run_id is not None, enforce terminal invariant.
# @POST: TranslationEvent row is created.

134
backend/src/plugins/translate/executor.py
View File

@@ -164,12 +164,102 @@ class TranslationExecutor:
# endregion execute_run
# region _fetch_source_rows [TYPE Function]
# @PURPOSE: Fetch source rows from the accepted preview session for this job.
# @PRE: job_id exists.
# @POST: Returns list of dicts with source data.
# @PURPOSE: Fetch full source dataset from Superset (via datasource) for full translation.
# @PRE: job_id exists. Job may have source_datasource_id for full fetch.
# @POST: Returns list of dicts with source data (all rows from the source datasource).
# @SIDE_EFFECT: Makes HTTP call to Superset chart data API when datasource is configured.
def _fetch_source_rows(self, job_id: str, run_id: str) -> List[Dict[str, Any]]:
with belief_scope("TranslationExecutor._fetch_source_rows"):
# Get the latest APPLIED preview session
job = self.db.query(TranslationJob).filter(TranslationJob.id == job_id).first()
# If source_datasource_id is configured, fetch ALL rows from the Superset chart data API
if job and job.source_datasource_id:
try:
logger.reason("Fetching full dataset from Superset datasource", {
"run_id": run_id,
"datasource_id": job.source_datasource_id,
"environment_id": job.environment_id,
})
# Determine environment
environments = self.config_manager.get_environments()
target_env_id = job.environment_id or job.source_dialect or ""
env_config = next(
(e for e in environments if e.id == target_env_id),
None,
)
if not env_config and environments:
env_config = environments[0]
if env_config:
from ...core.superset_client import SupersetClient
client = SupersetClient(env_config)
# Fetch dataset detail to build proper query context
dataset_detail = client.get_dataset_detail(int(job.source_datasource_id))
# Build query context (same approach as preview but without row_limit)
query_context = client.build_dataset_preview_query_context(
dataset_id=int(job.source_datasource_id),
dataset_record=dataset_detail,
template_params={},
effective_filters=[],
)
# Remove row_limit to get ALL rows; use result_type="samples"
queries = query_context.get("queries", [])
if queries:
queries[0].pop("row_limit", None)
queries[0].pop("result_type", None)
queries[0]["metrics"] = []
query_context["result_type"] = "samples"
form_data = query_context.get("form_data", {})
form_data.pop("query_mode", None)
response = client.network.request(
method="POST",
endpoint="/api/v1/chart/data",
data=json.dumps(query_context),
headers={"Content-Type": "application/json"},
)
# Extract rows
rows = self._extract_chart_data_rows(response)
if rows:
logger.reason(f"Fetched {len(rows)} rows from Superset datasource", {
"run_id": run_id,
})
# Map rows to source_rows format
source_rows = []
for idx, row in enumerate(rows):
source_data_dict = dict(row) if row else None
source_text = str(row.get(job.translation_column, "")) if job.translation_column else json.dumps(row)
source_rows.append({
"row_index": str(idx),
"source_text": source_text,
"approved_translation": None,
"source_object_name": f"Row {idx}",
"source_data": source_data_dict,
})
return source_rows
else:
logger.explore("Superset datasource returned no rows", {
"run_id": run_id,
"datasource_id": job.source_datasource_id,
})
else:
logger.explore("No environment config found for datasource fetch", {
"env_id": target_env_id,
})
except Exception as e:
logger.explore("Failed to fetch full dataset from Superset, falling back to preview", {
"run_id": run_id,
"error": str(e),
})
# Fall through to preview-based fetch
# Fallback: get the latest APPLIED preview session
session = (
self.db.query(TranslationPreviewSession)
.filter(
@@ -195,20 +285,48 @@ class TranslationExecutor:
source_rows = []
for rec in records:
source_data_dict = None
if hasattr(rec, "source_data") and rec.source_data:
source_data_dict = dict(rec.source_data)
source_rows.append({
"row_index": rec.source_object_id or "0",
"source_text": rec.source_sql or "",
"approved_translation": rec.target_sql if rec.status == "APPROVED" else None,
"source_object_name": rec.source_object_name or "",
"source_data": source_data_dict,
})
logger.reason(f"Fetched {len(source_rows)} source rows from preview", {
logger.reason(f"Fetched {len(source_rows)} source rows from preview fallback", {
"run_id": run_id,
"session_id": session.id,
})
return source_rows
# endregion _fetch_source_rows
# region _extract_chart_data_rows [TYPE Function]
# @PURPOSE: Extract data rows from Superset chart data API response.
# @POST: Returns list of dicts with column-value pairs.
@staticmethod
def _extract_chart_data_rows(response: Dict[str, Any]) -> List[Dict[str, Any]]:
result = response.get("result")
if isinstance(result, list):
for item in result:
if isinstance(item, dict):
data = item.get("data")
if isinstance(data, list) and data:
return data
if isinstance(result, dict):
data = result.get("data")
if isinstance(data, list) and data:
return data
data = response.get("data")
if isinstance(data, list) and data:
return data
if isinstance(result, list):
return result
return []
# endregion _extract_chart_data_rows
# region _process_batch [TYPE Function]
# @PURPOSE: Process a single batch: filter dict, build prompt, call LLM, persist records.
# @PRE: job and batch_rows are valid.
@@ -272,6 +390,7 @@ class TranslationExecutor:
source_object_type="table_row",
source_object_id=row.get("row_index"),
source_object_name=row.get("source_object_name", ""),
source_data=row.get("source_data"),
status="SUCCESS",
)
self.db.add(record)
@@ -398,6 +517,7 @@ class TranslationExecutor:
source_object_type="table_row",
source_object_id=row.get("row_index"),
source_object_name=row.get("source_object_name", ""),
source_data=row.get("source_data"),
status="FAILED",
error_message=f"LLM call failed after {retries} retries: {last_error}",
)
@@ -425,6 +545,7 @@ class TranslationExecutor:
source_object_type="table_row",
source_object_id=row.get("row_index"),
source_object_name=row.get("source_object_name", ""),
source_data=row.get("source_data"),
status="SKIPPED",
error_message=f"LLM parse failure: {e}",
)
@@ -456,6 +577,7 @@ class TranslationExecutor:
source_object_type="table_row",
source_object_id=row.get("row_index"),
source_object_name=row.get("source_object_name", ""),
source_data=row.get("source_data"),
status="SKIPPED",
error_message="NULL translation returned by LLM",
)
@@ -474,6 +596,7 @@ class TranslationExecutor:
source_object_type="table_row",
source_object_id=row.get("row_index"),
source_object_name=row.get("source_object_name", ""),
source_data=row.get("source_data"),
status="SKIPPED",
error_message="Empty translation returned by LLM",
)
@@ -490,6 +613,7 @@ class TranslationExecutor:
source_object_type="table_row",
source_object_id=row.get("row_index"),
source_object_name=row.get("source_object_name", ""),
source_data=row.get("source_data"),
status="SUCCESS",
)
self.db.add(record)

78
backend/src/plugins/translate/orchestrator.py
View File

@@ -63,7 +63,6 @@ class TranslationOrchestrator:
self._job: Optional[TranslationJob] = None
# region start_run [TYPE Function]
# @COMPLEXITY: 5
# @PURPOSE: Start a new translation run for a job.
# @PRE: job_id exists. For manual runs, there must be an accepted preview session.
# @POST: TranslationRun is created in PENDING status with hash fields and config snapshot. Events are recorded.
@@ -105,6 +104,7 @@ class TranslationOrchestrator:
"source_key_cols": job.source_key_cols,
"target_key_cols": job.target_key_cols,
"translation_column": job.translation_column,
"target_column": job.target_column,
"context_columns": job.context_columns,
"target_language": job.target_language,
"provider_id": job.provider_id,
@@ -263,7 +263,11 @@ class TranslationOrchestrator:
run.insert_status = insert_result.get("status")
run.superset_execution_id = str(insert_result.get("query_id") or "")
run.superset_execution_log = insert_result
run.status = "COMPLETED" if insert_result.get("status") == "success" else "COMPLETED"
# Preserve the translation-phase status. If the executor already
# marked the run FAILED (all LLM calls failed) we do NOT upgrade it
# to COMPLETED just because the insert phase ran.
if run.status != "FAILED":
run.status = "COMPLETED"
if insert_result.get("error_message"):
run.error_message = insert_result["error_message"]
run.completed_at = datetime.now(timezone.utc)
@@ -288,7 +292,9 @@ class TranslationOrchestrator:
)
self.db.commit()
self.db.refresh(run)
# Re-query run after commit — refresh may fail if the object was
# created in a different session or became detached during commit.
run = self.db.query(TranslationRun).filter(TranslationRun.id == run.id).first()
logger.reflect("Run execution complete", {
"run_id": run.id,
@@ -329,11 +335,20 @@ class TranslationOrchestrator:
"dialect": job.database_dialect or job.target_dialect,
})
# Build rows for SQL generation
# Determine effective target column for INSERT (defaults to translation_column)
effective_target = job.target_column or job.translation_column
# Build columns for SQL generation
columns = job.context_columns or []
# Always include translation_column (original text) if it's different from target
if job.translation_column and job.translation_column not in columns:
columns.append(job.translation_column)
# Add target_column separately if it differs from translation_column
if effective_target and effective_target != job.translation_column and effective_target not in columns:
columns.append(effective_target)
# Also include key columns if used for upsert
if job.target_key_cols:
for k in job.target_key_cols:
@@ -343,27 +358,57 @@ class TranslationOrchestrator:
rows_for_sql = []
for rec in records:
row_data = {}
source_data = rec.source_data or {}
# Context columns from source data
if job.context_columns:
for col in job.context_columns:
row_data[col] = ""
if job.translation_column:
row_data[job.translation_column] = rec.target_sql or ""
row_data[col] = source_data.get(col, "")
# Original text column (translation_column)
if job.translation_column and job.translation_column not in (job.target_key_cols or []):
# If target_column differs, keep the original value from source_data;
# otherwise use the translated value
if effective_target and effective_target != job.translation_column:
row_data[job.translation_column] = source_data.get(job.translation_column, "")
else:
row_data[job.translation_column] = rec.target_sql or ""
# Translated text goes into target_column (may be same as translation_column)
if effective_target:
row_data[effective_target] = rec.target_sql or ""
# Key columns from source data
if job.target_key_cols:
source_data = rec.source_data or {}
for k in job.target_key_cols:
# Use source_data value if available, fall back to empty string
row_data[k] = source_data.get(k, "")
rows_for_sql.append(row_data)
if not columns:
# Use target_sql as the sole column
columns = [job.translation_column or "translated_text"]
columns = [effective_target or "translated_text"]
rows_for_sql = [{columns[0]: rec.target_sql or ""} for rec in records]
# Resolve the real database backend engine from Superset
try:
env_id = job.environment_id or job.source_dialect or ""
executor = SupersetSqlLabExecutor(self.config_manager, env_id)
executor.resolve_database_id(
target_database_id=job.target_database_id,
)
real_backend = executor.get_database_backend()
except Exception as e:
logger.explore("Failed to resolve database backend, falling back to job dialet", {
"error": str(e),
})
real_backend = None
dialect = real_backend or job.database_dialect or job.target_dialect or "postgresql"
# Generate SQL
try:
sql, row_count = SQLGenerator.generate(
dialect=job.database_dialect or job.target_dialect or "postgresql",
dialect=dialect,
target_schema=job.target_schema,
target_table=job.target_table or "translated_data",
columns=columns,
@@ -375,19 +420,24 @@ class TranslationOrchestrator:
logger.explore("SQL generation failed", {"error": str(e)})
return {"status": "failed", "error_message": str(e), "query_id": None}
logger.reason("SQL generated with dialect", {
"dialect": dialect,
"real_backend": real_backend,
"job_database_dialect": job.database_dialect,
"job_target_dialect": job.target_dialect,
})
# Log insert phase start
self.event_log.log_event(
job_id=job.id,
run_id=run.id,
event_type="INSERT_PHASE_STARTED",
payload={"sql_length": len(sql), "row_count": row_count},
payload={"sql_length": len(sql), "row_count": row_count, "dialect": dialect},
created_by=self.current_user,
)
# Submit to Superset
try:
env_id = job.environment_id or job.source_dialect or ""
executor = SupersetSqlLabExecutor(self.config_manager, env_id)
result = executor.execute_and_poll(
sql=sql,
max_polls=30,

17
backend/src/plugins/translate/preview.py
View File

@@ -134,7 +134,6 @@ class TranslationPreview:
self.current_user = current_user
# region preview_rows [TYPE Function]
# @COMPLEXITY: 4
# @PURPOSE: Fetch sample rows from Superset dataset, send to LLM for translation, create preview session with records.
# @PRE: job_id exists and job has source_datasource_id, translation_column configured.
# @POST: Returns TranslationPreviewResponse with records, cost estimation, and persistent session.
@@ -290,6 +289,19 @@ class TranslationPreview:
status = "PENDING"
feedback = None
# Extract source_data: store original row key columns for upsert matching
source_row = meta.get("source_row", {})
source_data = None
if job.target_key_cols:
source_data = {
k: source_row.get(k)
for k in job.target_key_cols
if k in source_row
}
elif source_row:
# No key columns configured — store the full row as fallback
source_data = dict(source_row)
record = TranslationPreviewRecord(
id=str(uuid.uuid4()),
session_id=session.id,
@@ -298,6 +310,7 @@ class TranslationPreview:
source_object_type="table_row",
source_object_id=str(idx),
source_object_name=f"Row {idx + 1}",
source_data=source_data,
status=status,
feedback=feedback,
created_at=datetime.now(timezone.utc),
@@ -516,7 +529,6 @@ class TranslationPreview:
# endregion get_preview_session
# region _fetch_sample_rows [TYPE Function]
# @COMPLEXITY: 4
# @PURPOSE: Fetch sample rows from the Superset dataset for preview.
# @PRE: job has source_datasource_id and translation_column.
# @POST: Returns list of dicts with row data.
@@ -636,7 +648,6 @@ class TranslationPreview:
# endregion _extract_data_rows
# region _call_llm [TYPE Function]
# @COMPLEXITY: 4
# @PURPOSE: Call the configured LLM provider with the preview prompt.
# @PRE: job has a valid provider_id.
# @POST: Returns raw LLM response string.

5
backend/src/plugins/translate/service.py
View File

@@ -241,12 +241,14 @@ class TranslateJobService:
source_key_cols=payload.source_key_cols or [],
target_key_cols=payload.target_key_cols or [],
translation_column=payload.translation_column,
target_column=payload.target_column,
context_columns=payload.context_columns or [],
target_language=payload.target_language,
provider_id=payload.provider_id,
batch_size=payload.batch_size,
upsert_strategy=payload.upsert_strategy,
environment_id=payload.environment_id,
target_database_id=payload.target_database_id,
status="DRAFT",
created_by=self.current_user,
)
@@ -360,6 +362,7 @@ class TranslateJobService:
source_key_cols=source.source_key_cols,
target_key_cols=source.target_key_cols,
translation_column=source.translation_column,
target_column=source.target_column,
context_columns=source.context_columns,
target_language=source.target_language,
provider_id=source.provider_id,
@@ -461,6 +464,7 @@ def job_to_response(job: TranslationJob, dict_ids: Optional[List[str]] = None) -
source_key_cols=job.source_key_cols or [],
target_key_cols=job.target_key_cols or [],
translation_column=job.translation_column,
target_column=job.target_column,
context_columns=job.context_columns or [],
target_language=job.target_language,
provider_id=job.provider_id,
@@ -472,6 +476,7 @@ def job_to_response(job: TranslationJob, dict_ids: Optional[List[str]] = None) -
updated_at=job.updated_at,
dictionary_ids=dict_ids or [],
environment_id=job.environment_id,
target_database_id=job.target_database_id,
)
# #endregion job_to_response

15
backend/src/plugins/translate/sql_generator.py
View File

@@ -90,11 +90,16 @@ def _encode_sql_value(value: Any, dialect: Optional[str] = None) -> str:
if isinstance(value, bool):
return "TRUE" if value else "FALSE"
# For ClickHouse: try to normalize timestamp-like string values
if dialect in CLICKHOUSE_DIALECTS and isinstance(value, str) and value:
normalized = _normalize_timestamp_value(value)
if normalized:
return f"'{normalized}'"
# For ClickHouse: try to normalize timestamp-like values (both string and numeric)
if dialect in CLICKHOUSE_DIALECTS:
if isinstance(value, str) and value:
normalized = _normalize_timestamp_value(value)
if normalized:
return f"'{normalized}'"
elif isinstance(value, (int, float)):
normalized = _normalize_timestamp_value(value)
if normalized:
return f"'{normalized}'"
if isinstance(value, (int, float)):
return str(value)

146
backend/src/plugins/translate/superset_executor.py
View File

@@ -32,6 +32,7 @@ class SupersetSqlLabExecutor:
self.env_id = env_id
self._client: Optional[SupersetClient] = None
self._database_id: Optional[int] = None
self._database_backend: Optional[str] = None
# region _get_client [TYPE Function]
# @PURPOSE: Lazy-initialize SupersetClient for the configured environment.
@@ -53,15 +54,43 @@ class SupersetSqlLabExecutor:
# endregion _get_client
# region resolve_database_id [TYPE Function]
# @PURPOSE: Resolve the target database ID from the environment.
# @PURPOSE: Resolve the target database ID from the environment and fetch its backend engine.
# @PRE: database_name or database_id should be known.
# @POST: Returns database_id integer or raises ValueError.
# @SIDE_EFFECT: Fetches databases list from Superset.
def resolve_database_id(self, database_name: Optional[str] = None) -> int:
# @POST: Returns database_id integer or raises ValueError. Also sets self._database_backend.
# @SIDE_EFFECT: Fetches databases list from Superset; optionally fetches single DB for backend.
def resolve_database_id(
self,
database_name: Optional[str] = None,
target_database_id: Optional[str] = None,
) -> int:
with belief_scope("SupersetSqlLabExecutor.resolve_database_id"):
client = self._get_client()
# If a specific target_database_id is provided, use it directly
if target_database_id:
try:
db_id = int(target_database_id)
# Fetch full DB info to get backend
db_info = client.get_database(db_id)
result = db_info.get("result", db_info)
self._database_id = result.get("id", db_id)
self._database_backend = result.get("backend") or result.get("engine")
logger.reason("Resolved database ID by target_database_id", {
"target_database_id": target_database_id,
"database_id": self._database_id,
"backend": self._database_backend,
})
return self._database_id
except Exception as e:
logger.explore("Failed to resolve by target_database_id, falling back", {
"target_database_id": target_database_id,
"error": str(e),
})
# Fall through to default resolution
# Fetch full database list with backend info
_, databases = client.get_databases(
query={"columns": ["id", "database_name"]}
query={"columns": ["id", "database_name", "backend"]}
)
if not databases:
raise ValueError("No databases found in Superset environment")
@@ -70,9 +99,11 @@ class SupersetSqlLabExecutor:
for db in databases:
if db.get("database_name", "").lower() == database_name.lower():
self._database_id = db["id"]
self._database_backend = db.get("backend")
logger.reason("Resolved database ID by name", {
"database_name": database_name,
"database_id": self._database_id,
"backend": self._database_backend,
})
return self._database_id
raise ValueError(
@@ -81,13 +112,22 @@ class SupersetSqlLabExecutor:
# Default: use first database
self._database_id = databases[0]["id"]
self._database_backend = databases[0].get("backend")
logger.reason("Using default database", {
"database_name": databases[0].get("database_name"),
"database_id": self._database_id,
"backend": self._database_backend,
})
return self._database_id
# endregion resolve_database_id
# region get_database_backend [TYPE Function]
# @PURPOSE: Return the cached database backend/engine string.
# @POST: Returns backend string or None if not yet resolved.
def get_database_backend(self) -> Optional[str]:
return self._database_backend
# endregion get_database_backend
# region execute_sql [TYPE Function]
# @PURPOSE: Submit SQL to Superset SQL Lab and return execution tracking info.
# @PRE: sql is valid SQL string. database_id is a valid Superset DB ID.
@@ -117,27 +157,75 @@ class SupersetSqlLabExecutor:
"runAsync": run_async,
"schema": None,
"tab": "translation-insert",
"client_id": f"trl-{uuid.uuid4().hex[:7]}", # max 11 chars for Superset varchar(11)
}
try:
response = client.network.request(
method="POST",
endpoint="/api/v1/sqllab/execute/",
data=json.dumps(payload),
headers={"Content-Type": "application/json"},
)
except Exception as e:
logger.explore("SQL Lab execute failed", {"error": str(e)})
raise ValueError(f"Superset SQL Lab execute failed: {e}")
# Try multiple SQL Lab endpoints — some Superset versions use /sqllab/execute/,
# others use /sql_lab/execute/ (with underscore)
candidate_endpoints = ["/api/v1/sqllab/execute/", "/api/v1/sql_lab/execute/"]
response = None
last_error = None
for endpoint in candidate_endpoints:
try:
response = client.network.request(
method="POST",
endpoint=endpoint,
data=json.dumps(payload),
headers={"Content-Type": "application/json"},
)
if isinstance(response, dict) and response:
logger.reason(f"SQL Lab endpoint succeeded", extra={
"endpoint": endpoint,
})
break
except Exception as ep_err:
last_error = ep_err
logger.explore(f"SQL Lab endpoint failed, trying next", extra={
"error": str(ep_err),
"endpoint": endpoint,
})
response = None
# Parse response
if response is None:
error_msg = f"All SQL Lab endpoints failed. Last error: {last_error}"
logger.explore("SQL Lab execute failed", extra={"error": error_msg})
raise ValueError(f"Superset SQL Lab execute failed: {error_msg}")
# Parse response — try multiple known Superset response formats
result = response if isinstance(response, dict) else {}
query_id = result.get("query_id") or result.get("id")
# Superset may return query_id at top level, nested in "result", as "id", or in "query" block
query_id = (
result.get("query_id")
or result.get("id")
or (result.get("result") or {}).get("query_id")
or (result.get("result") or {}).get("id")
or (result.get("query") or {}).get("query_id")
or (result.get("query") or {}).get("id")
)
status = result.get("status", "unknown")
# Log full response for debugging if query_id is missing
if not query_id:
logger.explore("No query_id from SQL Lab execute", extra={
"database_id": db_id,
"status": status,
"raw_response_keys": list(result.keys()),
"raw_response_preview": json.dumps(result)[:2000],
})
logger.reason("SQL Lab execute response (no query_id)", extra={
"database_id": db_id,
"status": status,
"response_keys": list(result.keys()),
"has_result": "result" in result,
"has_query": "query" in result,
"raw_preview": json.dumps(result)[:2000],
})
logger.reason("SQL Lab execute response", {
"query_id": query_id,
"status": status,
"response_keys": list(result.keys()),
})
return {
@@ -174,7 +262,9 @@ class SupersetSqlLabExecutor:
method="GET",
endpoint=f"/api/v1/query/{query_id}",
)
result = response if isinstance(response, dict) else {}
raw = response if isinstance(response, dict) else {}
# Superset wraps query data in a nested "result" field
result = raw.get("result", raw)
status = result.get("status", "unknown")
state = result.get("state", result.get("status", ""))
@@ -255,9 +345,27 @@ class SupersetSqlLabExecutor:
exec_result = self.execute_sql(
sql=sql,
database_id=database_id,
run_async=True,
run_async=False, # Sync mode — Superset returns result directly
)
status = exec_result.get("status", "unknown")
# Sync mode: response already has the final status and data
if status == "success":
logger.reason("SQL execution completed (sync)", {
"query_id": exec_result.get("query_id"),
})
return {
"query_id": exec_result.get("query_id"),
"status": "success",
"state": "success",
"rows_affected": exec_result.get("raw_response", {}).get("query", {}).get("rows"),
"error_message": None,
"results": exec_result.get("raw_response", {}).get("data"),
"completed_on": exec_result.get("raw_response", {}).get("query", {}).get("endDttm"),
}
# For async mode (fallback): poll for completion
query_id = exec_result.get("query_id")
if not query_id:
logger.explore("No query_id from SQL Lab execute", {