feat: implement LLM table translation service with searchable datasource dropdown, i18n, sidebar, and RBAC
- Add backend plugin modules: preview, executor, orchestrator, events, sql_generator, superset_executor, dictionary, scheduler, metrics - Add API routes for translate jobs, runs, dictionaries, preview, schedule, metrics, corrections - Add ORM models (12 tables) and Pydantic schemas (15 DTOs) - Register translate router in app.py with RBAC permission guards - Add frontend pages: job list, job config, dictionary list/editor, history - Add 7 reusable Svelte 5 components: Preview, RunProgress, RunResult, TermCorrection, BulkCorrection, ScheduleConfig, MetricsDashboard - Add searchable datasource dropdown with Superset API integration - Add i18n (en/ru) with ~300 keys across common, config, jobs, dictionaries, preview namespaces - Add Translation sidebar category with Jobs/Dictionaries/History sub-items - Hide health monitor error toast via suppressToast API option - Add 69 backend tests and 44 frontend test files - Fix: SupersetClient env resolution (string -> Environment object) - Fix: Dataset detail API returns proper database dict - Fix: Database dialect extraction fallback when metadata incomplete
This commit is contained in:
855
backend/src/plugins/translate/orchestrator.py
Normal file
855
backend/src/plugins/translate/orchestrator.py
Normal file
@@ -0,0 +1,855 @@
|
||||
# [DEF:TranslationOrchestrator:Module]
|
||||
# @COMPLEXITY: 5
|
||||
# @SEMANTICS: translate, orchestrator, lifecycle, run
|
||||
# @PURPOSE: Run lifecycle coordination: validate preconditions, dispatch executor, generate SQL, submit to Superset, record events.
|
||||
# @LAYER: Domain
|
||||
# @RELATION: DEPENDS_ON -> [TranslationRun]
|
||||
# @RELATION: DEPENDS_ON -> [TranslationJob]
|
||||
# @RELATION: DEPENDS_ON -> [TranslationExecutor]
|
||||
# @RELATION: DEPENDS_ON -> [SQLGenerator]
|
||||
# @RELATION: DEPENDS_ON -> [SupersetSqlLabExecutor]
|
||||
# @RELATION: DEPENDS_ON -> [TranslationEventLog]
|
||||
# @RELATION: DEPENDS_ON -> [TranslationPreviewSession]
|
||||
# @PRE: Valid job and accepted preview (for manual runs). Superset and LLM are reachable.
|
||||
# @POST: Translation run is executed, SQL generated and submitted, events recorded.
|
||||
# @SIDE_EFFECT: Creates TranslationRun; creates batches and records; generates SQL; submits to Superset; records events.
|
||||
# @DATA_CONTRACT: Input[db, config_manager, current_user] -> Output[TranslationRun]
|
||||
# @INVARIANT: State transitions: PENDING -> RUNNING -> COMPLETED|FAILED|CANCELLED. Exactly one terminal event per run.
|
||||
# @RATIONALE: C5 orchestrator because preview, execution, event logging, and retry share state within single transaction boundary.
|
||||
# @REJECTED: Distributed actor model (Celery) — eventual-consistency challenges at current scale.
|
||||
# @REJECTED: stdout-only logging — lacks structured payload integrity; cannot enforce terminal-event invariant.
|
||||
|
||||
import json
|
||||
import time
|
||||
import uuid
|
||||
from datetime import datetime, timezone
|
||||
from typing import Any, Dict, List, Optional, Callable, Tuple
|
||||
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from ...core.logger import logger, belief_scope
|
||||
from ...core.config_manager import ConfigManager
|
||||
from ...models.translate import (
|
||||
TranslationJob,
|
||||
TranslationRun,
|
||||
TranslationBatch,
|
||||
TranslationRecord,
|
||||
TranslationPreviewSession,
|
||||
)
|
||||
from ...schemas.translate import TranslationRunResponse
|
||||
from .executor import TranslationExecutor
|
||||
from .sql_generator import SQLGenerator
|
||||
from .superset_executor import SupersetSqlLabExecutor
|
||||
from .events import TranslationEventLog
|
||||
from ..translate.service import TranslateJobService
|
||||
|
||||
|
||||
# [DEF:TranslationOrchestrator:Class]
|
||||
# @COMPLEXITY: 5
|
||||
# @PURPOSE: Coordinates full translation run lifecycle: validation, execution, SQL generation, Superset submission, event logging.
|
||||
# @PRE: DB session and config manager are available.
|
||||
# @POST: Runs are created, executed, and finalized with event records.
|
||||
# @SIDE_EFFECT: Creates/modifies TranslationRun, TranslationBatch, TranslationRecord, TranslationEvent rows.
|
||||
# @INVARIANT: State transitions: PENDING -> RUNNING -> COMPLETED|FAILED|CANCELLED.
|
||||
class TranslationOrchestrator:
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
db: Session,
|
||||
config_manager: ConfigManager,
|
||||
current_user: Optional[str] = None,
|
||||
):
|
||||
self.db = db
|
||||
self.config_manager = config_manager
|
||||
self.current_user = current_user
|
||||
self.event_log = TranslationEventLog(db)
|
||||
self._job: Optional[TranslationJob] = None
|
||||
|
||||
# [DEF:start_run: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.
|
||||
# @SIDE_EFFECT: DB writes.
|
||||
def start_run(
|
||||
self,
|
||||
job_id: str,
|
||||
is_scheduled: bool = False,
|
||||
trigger_type: Optional[str] = None,
|
||||
) -> TranslationRun:
|
||||
with belief_scope("TranslationOrchestrator.start_run"):
|
||||
# Load and validate job
|
||||
job = self.db.query(TranslationJob).filter(TranslationJob.id == job_id).first()
|
||||
if not job:
|
||||
raise ValueError(f"Translation job '{job_id}' not found")
|
||||
self._job = job
|
||||
|
||||
logger.reason("Starting translation run", {
|
||||
"job_id": job_id,
|
||||
"is_scheduled": is_scheduled,
|
||||
})
|
||||
|
||||
# Validate preconditions
|
||||
self._validate_preconditions(job, is_scheduled=is_scheduled)
|
||||
|
||||
# Compute hashes
|
||||
config_hash = self._compute_config_hash(job)
|
||||
dict_snapshot_hash = self._compute_dict_snapshot_hash(job_id)
|
||||
|
||||
# Build config snapshot
|
||||
config_snapshot = {
|
||||
"source_dialect": job.source_dialect,
|
||||
"target_dialect": job.target_dialect,
|
||||
"database_dialect": job.database_dialect,
|
||||
"source_datasource_id": job.source_datasource_id,
|
||||
"source_table": job.source_table,
|
||||
"target_schema": job.target_schema,
|
||||
"target_table": job.target_table,
|
||||
"source_key_cols": job.source_key_cols,
|
||||
"target_key_cols": job.target_key_cols,
|
||||
"translation_column": job.translation_column,
|
||||
"context_columns": job.context_columns,
|
||||
"target_language": job.target_language,
|
||||
"provider_id": job.provider_id,
|
||||
"batch_size": job.batch_size,
|
||||
"upsert_strategy": job.upsert_strategy,
|
||||
"dictionary_ids": self._compute_dict_snapshot_hash(job_id),
|
||||
}
|
||||
|
||||
# Compute key_hash from source_key_cols
|
||||
import hashlib
|
||||
key_hash_input = json.dumps({
|
||||
"source_key_cols": job.source_key_cols,
|
||||
"source_datasource_id": job.source_datasource_id,
|
||||
"source_table": job.source_table,
|
||||
}, sort_keys=True)
|
||||
key_hash = hashlib.sha256(key_hash_input.encode()).hexdigest()[:16]
|
||||
|
||||
# Create run record with all hash/snapshot fields
|
||||
run = TranslationRun(
|
||||
id=str(uuid.uuid4()),
|
||||
job_id=job_id,
|
||||
status="PENDING",
|
||||
trigger_type=trigger_type or ("scheduled" if is_scheduled else "manual"),
|
||||
config_snapshot=config_snapshot,
|
||||
key_hash=key_hash,
|
||||
config_hash=config_hash,
|
||||
dict_snapshot_hash=dict_snapshot_hash,
|
||||
created_by=self.current_user,
|
||||
created_at=datetime.now(timezone.utc),
|
||||
)
|
||||
self.db.add(run)
|
||||
self.db.flush()
|
||||
|
||||
# Record event
|
||||
self.event_log.log_event(
|
||||
job_id=job_id,
|
||||
run_id=run.id,
|
||||
event_type="RUN_STARTED",
|
||||
payload={
|
||||
"is_scheduled": is_scheduled,
|
||||
"trigger_type": run.trigger_type,
|
||||
"config_hash": config_hash,
|
||||
"dict_snapshot_hash": dict_snapshot_hash,
|
||||
"key_hash": key_hash,
|
||||
},
|
||||
created_by=self.current_user,
|
||||
)
|
||||
|
||||
self.db.commit()
|
||||
self.db.refresh(run)
|
||||
|
||||
logger.reflect("Run created", {
|
||||
"run_id": run.id,
|
||||
"job_id": job_id,
|
||||
"status": run.status,
|
||||
"trigger_type": run.trigger_type,
|
||||
})
|
||||
return run
|
||||
# [/DEF:start_run:Function]
|
||||
|
||||
# [DEF:execute_run:Function]
|
||||
# @PURPOSE: Execute a translation run: dispatch executor, generate SQL, submit to Superset.
|
||||
# @PRE: run is in PENDING status.
|
||||
# @POST: Run is executed, SQL generated, Superset submission attempted.
|
||||
# @SIDE_EFFECT: LLM calls, DB writes, Superset API calls.
|
||||
def execute_run(
|
||||
self,
|
||||
run: TranslationRun,
|
||||
on_batch_progress: Optional[Callable[[str, int, int, int, int], None]] = None,
|
||||
skip_insert: bool = False,
|
||||
) -> TranslationRun:
|
||||
with belief_scope("TranslationOrchestrator.execute_run"):
|
||||
job = self.db.query(TranslationJob).filter(TranslationJob.id == run.job_id).first()
|
||||
if not job:
|
||||
raise ValueError(f"Job '{run.job_id}' not found")
|
||||
self._job = job
|
||||
|
||||
if run.status != "PENDING":
|
||||
raise ValueError(
|
||||
f"Cannot execute run in status '{run.status}'. "
|
||||
f"Run must be in PENDING status."
|
||||
)
|
||||
|
||||
logger.reason("Executing run", {
|
||||
"run_id": run.id,
|
||||
"job_id": job.id,
|
||||
"skip_insert": skip_insert,
|
||||
})
|
||||
|
||||
# Record translation phase start
|
||||
self.event_log.log_event(
|
||||
job_id=job.id,
|
||||
run_id=run.id,
|
||||
event_type="TRANSLATION_PHASE_STARTED",
|
||||
payload={},
|
||||
created_by=self.current_user,
|
||||
)
|
||||
|
||||
# Dispatch executor
|
||||
executor = TranslationExecutor(
|
||||
self.db, self.config_manager, self.current_user,
|
||||
on_batch_progress=on_batch_progress,
|
||||
)
|
||||
try:
|
||||
run = executor.execute_run(run, llm_progress_callback=None)
|
||||
except Exception as e:
|
||||
logger.explore("Translation execution failed", {
|
||||
"run_id": run.id,
|
||||
"error": str(e),
|
||||
})
|
||||
run.status = "FAILED"
|
||||
run.error_message = f"Translation execution failed: {e}"
|
||||
run.completed_at = datetime.now(timezone.utc)
|
||||
self.db.flush()
|
||||
self.event_log.log_event(
|
||||
job_id=job.id,
|
||||
run_id=run.id,
|
||||
event_type="RUN_FAILED",
|
||||
payload={"error": str(e), "phase": "translation"},
|
||||
created_by=self.current_user,
|
||||
)
|
||||
self.db.commit()
|
||||
return run
|
||||
|
||||
# Record translation phase complete
|
||||
self.event_log.log_event(
|
||||
job_id=job.id,
|
||||
run_id=run.id,
|
||||
event_type="TRANSLATION_PHASE_COMPLETED",
|
||||
payload={
|
||||
"total": run.total_records,
|
||||
"successful": run.successful_records,
|
||||
"failed": run.failed_records,
|
||||
"skipped": run.skipped_records,
|
||||
},
|
||||
created_by=self.current_user,
|
||||
)
|
||||
|
||||
# Skip insert phase if requested (e.g., for preview-only execution)
|
||||
if skip_insert:
|
||||
run.status = "COMPLETED"
|
||||
run.completed_at = datetime.now(timezone.utc)
|
||||
self.db.flush()
|
||||
self.event_log.log_event(
|
||||
job_id=job.id,
|
||||
run_id=run.id,
|
||||
event_type="RUN_COMPLETED",
|
||||
payload={"skip_insert": True},
|
||||
created_by=self.current_user,
|
||||
)
|
||||
self.db.commit()
|
||||
return run
|
||||
|
||||
# Generate SQL and submit to Superset
|
||||
insert_result = self._generate_and_insert_sql(job, run)
|
||||
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"
|
||||
if insert_result.get("error_message"):
|
||||
run.error_message = insert_result["error_message"]
|
||||
run.completed_at = datetime.now(timezone.utc)
|
||||
self.db.flush()
|
||||
|
||||
# Record terminal event
|
||||
terminal_event = "RUN_COMPLETED"
|
||||
self.event_log.log_event(
|
||||
job_id=job.id,
|
||||
run_id=run.id,
|
||||
event_type=terminal_event,
|
||||
payload={
|
||||
"insert_status": insert_result.get("status"),
|
||||
"query_id": insert_result.get("query_id"),
|
||||
"rows_affected": insert_result.get("rows_affected"),
|
||||
"total_records": run.total_records,
|
||||
"successful": run.successful_records,
|
||||
"failed": run.failed_records,
|
||||
"skipped": run.skipped_records,
|
||||
},
|
||||
created_by=self.current_user,
|
||||
)
|
||||
|
||||
self.db.commit()
|
||||
self.db.refresh(run)
|
||||
|
||||
logger.reflect("Run execution complete", {
|
||||
"run_id": run.id,
|
||||
"status": run.status,
|
||||
"insert_status": run.insert_status,
|
||||
})
|
||||
return run
|
||||
# [/DEF:execute_run:Function]
|
||||
|
||||
# [DEF:_generate_and_insert_sql:Function]
|
||||
# @PURPOSE: Generate INSERT SQL from successful records and submit to Superset SQL Lab.
|
||||
# @PRE: job has target table configured. run has successful records.
|
||||
# @POST: SQL is generated and submitted. Returns execution result.
|
||||
# @SIDE_EFFECT: Superset API call.
|
||||
def _generate_and_insert_sql(
|
||||
self,
|
||||
job: TranslationJob,
|
||||
run: TranslationRun,
|
||||
) -> Dict[str, Any]:
|
||||
with belief_scope("TranslationOrchestrator._generate_and_insert_sql"):
|
||||
# Fetch successful records
|
||||
records = (
|
||||
self.db.query(TranslationRecord)
|
||||
.filter(
|
||||
TranslationRecord.run_id == run.id,
|
||||
TranslationRecord.status == "SUCCESS",
|
||||
TranslationRecord.target_sql.isnot(None),
|
||||
)
|
||||
.all()
|
||||
)
|
||||
|
||||
if not records:
|
||||
logger.reason("No successful records to insert", {"run_id": run.id})
|
||||
return {"status": "skipped", "reason": "no_records", "query_id": None}
|
||||
|
||||
logger.reason(f"Generating SQL for {len(records)} records", {
|
||||
"run_id": run.id,
|
||||
"dialect": job.database_dialect or job.target_dialect,
|
||||
})
|
||||
|
||||
# Build rows for SQL generation
|
||||
columns = job.context_columns or []
|
||||
if job.translation_column and job.translation_column not in columns:
|
||||
columns.append(job.translation_column)
|
||||
|
||||
# Also include key columns if used for upsert
|
||||
if job.target_key_cols:
|
||||
for k in job.target_key_cols:
|
||||
if k not in columns:
|
||||
columns.append(k)
|
||||
|
||||
rows_for_sql = []
|
||||
for rec in records:
|
||||
row_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 ""
|
||||
if job.target_key_cols:
|
||||
for k in job.target_key_cols:
|
||||
row_data[k] = ""
|
||||
rows_for_sql.append(row_data)
|
||||
|
||||
if not columns:
|
||||
# Use target_sql as the sole column
|
||||
columns = [job.translation_column or "translated_text"]
|
||||
rows_for_sql = [{columns[0]: rec.target_sql or ""} for rec in records]
|
||||
|
||||
# Generate SQL
|
||||
try:
|
||||
sql, row_count = SQLGenerator.generate(
|
||||
dialect=job.database_dialect or job.target_dialect or "postgresql",
|
||||
target_schema=job.target_schema,
|
||||
target_table=job.target_table or "translated_data",
|
||||
columns=columns,
|
||||
rows=rows_for_sql,
|
||||
key_columns=job.target_key_cols,
|
||||
upsert_strategy=job.upsert_strategy or "MERGE",
|
||||
)
|
||||
except ValueError as e:
|
||||
logger.explore("SQL generation failed", {"error": str(e)})
|
||||
return {"status": "failed", "error_message": str(e), "query_id": None}
|
||||
|
||||
# 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},
|
||||
created_by=self.current_user,
|
||||
)
|
||||
|
||||
# Submit to Superset
|
||||
try:
|
||||
env_id = job.source_dialect or ""
|
||||
executor = SupersetSqlLabExecutor(self.config_manager, env_id)
|
||||
result = executor.execute_and_poll(
|
||||
sql=sql,
|
||||
max_polls=30,
|
||||
poll_interval_seconds=2.0,
|
||||
)
|
||||
except Exception as e:
|
||||
logger.explore("Superset SQL submission failed", {
|
||||
"run_id": run.id,
|
||||
"error": str(e),
|
||||
})
|
||||
result = {"status": "failed", "error_message": str(e), "query_id": None}
|
||||
|
||||
# Log insert phase complete
|
||||
self.event_log.log_event(
|
||||
job_id=job.id,
|
||||
run_id=run.id,
|
||||
event_type="INSERT_PHASE_COMPLETED",
|
||||
payload=result,
|
||||
created_by=self.current_user,
|
||||
)
|
||||
|
||||
return result
|
||||
# [/DEF:_generate_and_insert_sql:Function]
|
||||
|
||||
# [DEF:_validate_preconditions:Function]
|
||||
# @PURPOSE: Validate preconditions before starting a run.
|
||||
# @PRE: None.
|
||||
# @POST: Raises ValueError if preconditions are not met.
|
||||
# @SIDE_EFFECT: None.
|
||||
def _validate_preconditions(
|
||||
self,
|
||||
job: TranslationJob,
|
||||
is_scheduled: bool = False,
|
||||
) -> None:
|
||||
with belief_scope("TranslationOrchestrator._validate_preconditions"):
|
||||
# Job must be in valid status
|
||||
if job.status in ("DRAFT",):
|
||||
raise ValueError(
|
||||
f"Cannot run job '{job.id}' in status '{job.status}'. "
|
||||
f"Job must be READY, ACTIVE, or COMPLETED."
|
||||
)
|
||||
|
||||
# Must have target table configured
|
||||
if not job.target_table:
|
||||
raise ValueError(
|
||||
f"Job '{job.id}' has no target table configured. "
|
||||
"Configure a target table before running."
|
||||
)
|
||||
|
||||
# Must have LLM provider configured
|
||||
if not job.provider_id:
|
||||
raise ValueError(
|
||||
f"Job '{job.id}' has no LLM provider configured. "
|
||||
"Select an LLM provider before running."
|
||||
)
|
||||
|
||||
# Must have a translation column
|
||||
if not job.translation_column:
|
||||
raise ValueError(
|
||||
f"Job '{job.id}' has no translation column configured. "
|
||||
"Select a translation column before running."
|
||||
)
|
||||
|
||||
# For manual runs, must have accepted preview
|
||||
if not is_scheduled:
|
||||
accepted_session = (
|
||||
self.db.query(TranslationPreviewSession)
|
||||
.filter(
|
||||
TranslationPreviewSession.job_id == job.id,
|
||||
TranslationPreviewSession.status == "APPLIED",
|
||||
)
|
||||
.order_by(TranslationPreviewSession.created_at.desc())
|
||||
.first()
|
||||
)
|
||||
if not accepted_session:
|
||||
raise ValueError(
|
||||
f"Job '{job.id}' has no accepted preview session. "
|
||||
"Run and accept a preview before executing a manual translation run."
|
||||
)
|
||||
|
||||
logger.reason("Preconditions validated", {
|
||||
"job_id": job.id,
|
||||
"is_scheduled": is_scheduled,
|
||||
})
|
||||
# [/DEF:_validate_preconditions:Function]
|
||||
|
||||
# [DEF:retry_failed_batches:Function]
|
||||
# @PURPOSE: Retry failed batches in a run.
|
||||
# @PRE: run exists and has failed batches.
|
||||
# @POST: Failed batches are re-processed.
|
||||
# @SIDE_EFFECT: LLM calls, DB writes.
|
||||
def retry_failed_batches(
|
||||
self,
|
||||
run_id: str,
|
||||
) -> TranslationRun:
|
||||
with belief_scope("TranslationOrchestrator.retry_failed_batches"):
|
||||
run = self.db.query(TranslationRun).filter(TranslationRun.id == run_id).first()
|
||||
if not run:
|
||||
raise ValueError(f"Run '{run_id}' not found")
|
||||
|
||||
job = self.db.query(TranslationJob).filter(TranslationJob.id == run.job_id).first()
|
||||
if not job:
|
||||
raise ValueError(f"Job '{run.job_id}' not found")
|
||||
self._job = job
|
||||
|
||||
# Find failed batches
|
||||
failed_batches = (
|
||||
self.db.query(TranslationBatch)
|
||||
.filter(
|
||||
TranslationBatch.run_id == run_id,
|
||||
TranslationBatch.status.in_(["FAILED", "COMPLETED_WITH_ERRORS"]),
|
||||
)
|
||||
.all()
|
||||
)
|
||||
|
||||
if not failed_batches:
|
||||
raise ValueError(f"No failed batches found for run '{run_id}'")
|
||||
|
||||
logger.reason("Retrying failed batches", {
|
||||
"run_id": run_id,
|
||||
"batch_count": len(failed_batches),
|
||||
})
|
||||
|
||||
self.event_log.log_event(
|
||||
job_id=job.id,
|
||||
run_id=run.id,
|
||||
event_type="RUN_RETRYING",
|
||||
payload={"batch_count": len(failed_batches)},
|
||||
created_by=self.current_user,
|
||||
)
|
||||
|
||||
# Re-process each failed batch
|
||||
executor = TranslationExecutor(self.db, self.config_manager, self.current_user)
|
||||
run.status = "RUNNING"
|
||||
self.db.flush()
|
||||
|
||||
for batch in failed_batches:
|
||||
# Fetch failed records for this batch
|
||||
failed_records = (
|
||||
self.db.query(TranslationRecord)
|
||||
.filter(
|
||||
TranslationRecord.batch_id == batch.id,
|
||||
TranslationRecord.status == "FAILED",
|
||||
)
|
||||
.all()
|
||||
)
|
||||
|
||||
if not failed_records:
|
||||
continue
|
||||
|
||||
# Build rows from failed records
|
||||
retry_rows = []
|
||||
for rec in failed_records:
|
||||
retry_rows.append({
|
||||
"row_index": rec.source_object_id or "0",
|
||||
"source_text": rec.source_sql or "",
|
||||
"approved_translation": None,
|
||||
"source_object_name": rec.source_object_name or "",
|
||||
})
|
||||
|
||||
# Process retry batch
|
||||
result = executor._process_batch(
|
||||
job=job,
|
||||
run_id=run_id,
|
||||
batch_index=batch.batch_index,
|
||||
batch_rows=retry_rows,
|
||||
)
|
||||
|
||||
# Update run stats
|
||||
run.successful_records = (run.successful_records or 0) + result["successful"]
|
||||
run.failed_records = (run.failed_records or 0) + result["failed"]
|
||||
run.skipped_records = (run.skipped_records or 0) + result["skipped"]
|
||||
self.db.flush()
|
||||
|
||||
# Update run status
|
||||
if run.failed_records == 0:
|
||||
run.status = "COMPLETED"
|
||||
elif run.successful_records == 0:
|
||||
run.status = "FAILED"
|
||||
else:
|
||||
run.status = "COMPLETED"
|
||||
|
||||
run.completed_at = datetime.now(timezone.utc)
|
||||
self.db.flush()
|
||||
|
||||
self.event_log.log_event(
|
||||
job_id=job.id,
|
||||
run_id=run.id,
|
||||
event_type="RUN_COMPLETED",
|
||||
payload={"retry": True},
|
||||
created_by=self.current_user,
|
||||
)
|
||||
|
||||
self.db.commit()
|
||||
logger.reflect("Retry complete", {
|
||||
"run_id": run_id,
|
||||
"status": run.status,
|
||||
})
|
||||
return run
|
||||
# [/DEF:retry_failed_batches:Function]
|
||||
|
||||
# [DEF:retry_insert:Function]
|
||||
# @PURPOSE: Retry the SQL insert phase for a completed run.
|
||||
# @PRE: run exists and has successful records.
|
||||
# @POST: SQL is regenerated and re-submitted to Superset.
|
||||
# @SIDE_EFFECT: Superset API call.
|
||||
def retry_insert(
|
||||
self,
|
||||
run_id: str,
|
||||
) -> TranslationRun:
|
||||
with belief_scope("TranslationOrchestrator.retry_insert"):
|
||||
run = self.db.query(TranslationRun).filter(TranslationRun.id == run_id).first()
|
||||
if not run:
|
||||
raise ValueError(f"Run '{run_id}' not found")
|
||||
|
||||
job = self.db.query(TranslationJob).filter(TranslationJob.id == run.job_id).first()
|
||||
if not job:
|
||||
raise ValueError(f"Job '{run.job_id}' not found")
|
||||
self._job = job
|
||||
|
||||
logger.reason("Retrying insert phase", {
|
||||
"run_id": run_id,
|
||||
})
|
||||
|
||||
self.event_log.log_event(
|
||||
job_id=job.id,
|
||||
run_id=run.id,
|
||||
event_type="RUN_RETRY_INSERT",
|
||||
payload={},
|
||||
created_by=self.current_user,
|
||||
)
|
||||
|
||||
# Regenerate SQL and submit
|
||||
insert_result = self._generate_and_insert_sql(job, run)
|
||||
|
||||
# Update run
|
||||
run.insert_status = insert_result.get("status")
|
||||
run.superset_execution_id = str(insert_result.get("query_id") or "")
|
||||
if insert_result.get("error_message"):
|
||||
run.error_message = insert_result["error_message"]
|
||||
self.db.flush()
|
||||
self.db.commit()
|
||||
self.db.refresh(run)
|
||||
|
||||
logger.reflect("Insert retry complete", {
|
||||
"run_id": run_id,
|
||||
"insert_status": run.insert_status,
|
||||
})
|
||||
return run
|
||||
# [/DEF:retry_insert:Function]
|
||||
|
||||
# [DEF:cancel_run:Function]
|
||||
# @PURPOSE: Cancel a running translation.
|
||||
# @PRE: run is in PENDING or RUNNING status.
|
||||
# @POST: Run status is set to CANCELLED.
|
||||
# @SIDE_EFFECT: DB write; records event.
|
||||
def cancel_run(self, run_id: str) -> TranslationRun:
|
||||
with belief_scope("TranslationOrchestrator.cancel_run"):
|
||||
run = self.db.query(TranslationRun).filter(TranslationRun.id == run_id).first()
|
||||
if not run:
|
||||
raise ValueError(f"Run '{run_id}' not found")
|
||||
|
||||
if run.status not in ("PENDING", "RUNNING"):
|
||||
raise ValueError(
|
||||
f"Cannot cancel run in status '{run.status}'. "
|
||||
f"Only PENDING or RUNNING runs can be cancelled."
|
||||
)
|
||||
|
||||
run.status = "CANCELLED"
|
||||
run.completed_at = datetime.now(timezone.utc)
|
||||
self.db.flush()
|
||||
|
||||
self.event_log.log_event(
|
||||
job_id=run.job_id,
|
||||
run_id=run.id,
|
||||
event_type="RUN_CANCELLED",
|
||||
payload={},
|
||||
created_by=self.current_user,
|
||||
)
|
||||
|
||||
self.db.commit()
|
||||
self.db.refresh(run)
|
||||
|
||||
logger.reflect("Run cancelled", {
|
||||
"run_id": run_id,
|
||||
})
|
||||
return run
|
||||
# [/DEF:cancel_run:Function]
|
||||
|
||||
# [DEF:get_run_status:Function]
|
||||
# @PURPOSE: Get run status with statistics.
|
||||
# @PRE: run_id exists.
|
||||
# @POST: Returns dict with run details.
|
||||
def get_run_status(self, run_id: str) -> Dict[str, Any]:
|
||||
with belief_scope("TranslationOrchestrator.get_run_status"):
|
||||
run = self.db.query(TranslationRun).filter(TranslationRun.id == run_id).first()
|
||||
if not run:
|
||||
raise ValueError(f"Run '{run_id}' not found")
|
||||
|
||||
# Count batches
|
||||
batch_count = (
|
||||
self.db.query(TranslationBatch)
|
||||
.filter(TranslationBatch.run_id == run_id)
|
||||
.count()
|
||||
)
|
||||
|
||||
# Get event summary
|
||||
event_summary = self.event_log.get_run_event_summary(run_id)
|
||||
|
||||
return {
|
||||
"id": run.id,
|
||||
"job_id": run.job_id,
|
||||
"status": run.status,
|
||||
"started_at": run.started_at.isoformat() if run.started_at else None,
|
||||
"completed_at": run.completed_at.isoformat() if run.completed_at else None,
|
||||
"error_message": run.error_message,
|
||||
"total_records": run.total_records or 0,
|
||||
"successful_records": run.successful_records or 0,
|
||||
"failed_records": run.failed_records or 0,
|
||||
"skipped_records": run.skipped_records or 0,
|
||||
"insert_status": run.insert_status,
|
||||
"superset_execution_id": run.superset_execution_id,
|
||||
"batch_count": batch_count,
|
||||
"event_invariants": {
|
||||
"has_run_started": event_summary["has_run_started"],
|
||||
"terminal_event_count": event_summary["terminal_event_count"],
|
||||
"invariant_valid": event_summary["invariant_valid"],
|
||||
},
|
||||
"created_by": run.created_by,
|
||||
"created_at": run.created_at.isoformat() if run.created_at else None,
|
||||
}
|
||||
# [/DEF:get_run_status:Function]
|
||||
|
||||
# [DEF:get_run_records:Function]
|
||||
# @PURPOSE: Get paginated records for a run.
|
||||
# @PRE: run_id exists.
|
||||
# @POST: Returns dict with records and pagination info.
|
||||
def get_run_records(
|
||||
self,
|
||||
run_id: str,
|
||||
page: int = 1,
|
||||
page_size: int = 50,
|
||||
status_filter: Optional[str] = None,
|
||||
) -> Dict[str, Any]:
|
||||
with belief_scope("TranslationOrchestrator.get_run_records"):
|
||||
query = self.db.query(TranslationRecord).filter(
|
||||
TranslationRecord.run_id == run_id
|
||||
)
|
||||
|
||||
if status_filter:
|
||||
query = query.filter(TranslationRecord.status == status_filter)
|
||||
|
||||
total = query.count()
|
||||
offset = (page - 1) * page_size
|
||||
records = (
|
||||
query.order_by(TranslationRecord.created_at.desc())
|
||||
.offset(offset)
|
||||
.limit(page_size)
|
||||
.all()
|
||||
)
|
||||
|
||||
return {
|
||||
"items": [
|
||||
{
|
||||
"id": r.id,
|
||||
"batch_id": r.batch_id,
|
||||
"source_sql": r.source_sql,
|
||||
"target_sql": r.target_sql,
|
||||
"source_object_type": r.source_object_type,
|
||||
"source_object_id": r.source_object_id,
|
||||
"source_object_name": r.source_object_name,
|
||||
"status": r.status,
|
||||
"error_message": r.error_message,
|
||||
"created_at": r.created_at.isoformat() if r.created_at else None,
|
||||
}
|
||||
for r in records
|
||||
],
|
||||
"total": total,
|
||||
"page": page,
|
||||
"page_size": page_size,
|
||||
"status_filter": status_filter,
|
||||
}
|
||||
# [/DEF:get_run_records:Function]
|
||||
|
||||
# [DEF:get_run_history:Function]
|
||||
# @PURPOSE: Get run history for a job.
|
||||
# @PRE: job_id exists.
|
||||
# @POST: Returns list of runs.
|
||||
def get_run_history(
|
||||
self,
|
||||
job_id: str,
|
||||
page: int = 1,
|
||||
page_size: int = 20,
|
||||
) -> Tuple[int, List[Dict[str, Any]]]:
|
||||
with belief_scope("TranslationOrchestrator.get_run_history"):
|
||||
query = self.db.query(TranslationRun).filter(
|
||||
TranslationRun.job_id == job_id
|
||||
)
|
||||
total = query.count()
|
||||
runs = (
|
||||
query.order_by(TranslationRun.created_at.desc())
|
||||
.offset((page - 1) * page_size)
|
||||
.limit(page_size)
|
||||
.all()
|
||||
)
|
||||
|
||||
return total, [
|
||||
{
|
||||
"id": r.id,
|
||||
"job_id": r.job_id,
|
||||
"status": r.status,
|
||||
"started_at": r.started_at.isoformat() if r.started_at else None,
|
||||
"completed_at": r.completed_at.isoformat() if r.completed_at else None,
|
||||
"error_message": r.error_message,
|
||||
"total_records": r.total_records or 0,
|
||||
"successful_records": r.successful_records or 0,
|
||||
"failed_records": r.failed_records or 0,
|
||||
"skipped_records": r.skipped_records or 0,
|
||||
"insert_status": r.insert_status,
|
||||
"superset_execution_id": r.superset_execution_id,
|
||||
"created_by": r.created_by,
|
||||
"created_at": r.created_at.isoformat() if r.created_at else None,
|
||||
}
|
||||
for r in runs
|
||||
]
|
||||
# [/DEF:get_run_history:Function]
|
||||
|
||||
# [DEF:_compute_config_hash:Function]
|
||||
# @PURPOSE: Compute a hash of the job's current configuration for snapshot comparison.
|
||||
@staticmethod
|
||||
def _compute_config_hash(job: TranslationJob) -> str:
|
||||
import hashlib
|
||||
config_str = json.dumps({
|
||||
"source_dialect": job.source_dialect,
|
||||
"target_dialect": job.target_dialect,
|
||||
"source_datasource_id": job.source_datasource_id,
|
||||
"translation_column": job.translation_column,
|
||||
"context_columns": job.context_columns,
|
||||
"target_language": job.target_language,
|
||||
"provider_id": job.provider_id,
|
||||
"batch_size": job.batch_size,
|
||||
"upsert_strategy": job.upsert_strategy,
|
||||
}, sort_keys=True)
|
||||
return hashlib.sha256(config_str.encode()).hexdigest()[:16]
|
||||
# [/DEF:_compute_config_hash:Function]
|
||||
|
||||
# [DEF:_compute_dict_snapshot_hash:Function]
|
||||
# @PURPOSE: Compute a hash of dictionary state for snapshot comparison.
|
||||
def _compute_dict_snapshot_hash(self, job_id: str) -> str:
|
||||
import hashlib
|
||||
from ...models.translate import TranslationJobDictionary
|
||||
dict_links = (
|
||||
self.db.query(TranslationJobDictionary)
|
||||
.filter(TranslationJobDictionary.job_id == job_id)
|
||||
.all()
|
||||
)
|
||||
dict_ids = sorted([dl.dictionary_id for dl in dict_links])
|
||||
hash_input = ",".join(dict_ids)
|
||||
return hashlib.sha256(hash_input.encode()).hexdigest()[:16]
|
||||
# [/DEF:_compute_dict_snapshot_hash:Function]
|
||||
|
||||
|
||||
# [/DEF:TranslationOrchestrator:Class]
|
||||
# [/DEF:TranslationOrchestrator:Module]
|
||||
Reference in New Issue
Block a user