diff --git a/backend/src/api/routes/translate/_run_routes.py b/backend/src/api/routes/translate/_run_routes.py index df0d7dda..3adf6a5a 100644 --- a/backend/src/api/routes/translate/_run_routes.py +++ b/backend/src/api/routes/translate/_run_routes.py @@ -7,6 +7,7 @@ from datetime import UTC, datetime +import asyncio from fastapi import Depends, HTTPException, Query, status from sqlalchemy.orm import Session @@ -47,16 +48,12 @@ def run_translation( orch = TranslationOrchestrator(db, config_manager, current_user.username) run = orch.start_run(job_id=job_id, is_scheduled=False, full_translation=full_translation) # The request-scoped db session will be closed after this handler returns. - # The background thread must use its OWN session to avoid operating on a - # closed or expunged session. - import threading - - # Capture request trace_id for background thread correlation + # The background task must use its OWN session. _request_trace_id = get_trace_id() - def _background_execute(): + async def _background_execute(): from ....models.translate import TranslationRun as TRModel - # Preserve request's trace_id for log correlation across the thread boundary + # Preserve request's trace_id for log correlation if _request_trace_id: set_trace_id(_request_trace_id) else: @@ -68,8 +65,6 @@ def run_translation( bg_db, config_manager, current_user.username if current_user else None, ) - # Re-fetch the run within the background session to get a fresh, - # attached object bg_run = bg_db.query(TRModel).filter(TRModel.id == run.id).first() if bg_run is None: logger.explore( @@ -78,55 +73,30 @@ def run_translation( ) return - # execute_run internally calls self.db.commit() - bg_orch.execute_run(bg_run) + await bg_orch.execute_run(bg_run) - # ----- VERIFICATION ----- - # After execute_run the run object from bg_db is detached after - # commit. Re-query to verify the status was persisted in the - # database and is a terminal state. check_run = bg_db.query(TRModel).filter(TRModel.id == run.id).first() if check_run and check_run.status in ("COMPLETED", "FAILED", "CANCELLED"): logger.reason( "Background execute verified", - extra={ - "src": "translate_routes", - "run_id": run.id, - "status": check_run.status, - }, + extra={"src": "translate_routes", "run_id": run.id, "status": check_run.status}, ) else: - # execute_run appeared to succeed but the run is still in a - # non-terminal state — manually fail it so the frontend - # polling can detect the terminal state and stop spinning. actual_status = check_run.status if check_run else "NOT_FOUND" logger.explore( "Background execute: run not in terminal state after commit", - extra={ - "src": "translate_routes", - "run_id": run.id, - "status": actual_status, - }, + extra={"src": "translate_routes", "run_id": run.id, "status": actual_status}, ) if check_run: check_run.status = "FAILED" - check_run.error_message = ( - "Background execution did not reach terminal state " - f"(was: {actual_status})" - ) + check_run.error_message = f"Background execution did not reach terminal state (was: {actual_status})" bg_db.commit() except Exception as bg_err: logger.explore( "Background execute failed", - extra={ - "src": "translate_routes", - "run_id": run.id, - "error": str(bg_err), - }, + extra={"src": "translate_routes", "run_id": run.id, "error": str(bg_err)}, ) - # Mark the run as FAILED so the frontend polling can detect a - # terminal state and stop spinning. try: fb_db = SessionLocal() try: @@ -136,29 +106,18 @@ def run_translation( fb_run.error_message = f"Background execute error: {bg_err}" fb_run.completed_at = datetime.now(UTC) fb_db.commit() - logger.reason( - "Background execute: run marked FAILED", - extra={"src": "translate_routes", "run_id": run.id}, - ) + logger.reason("Background execute: run marked FAILED", + extra={"src": "translate_routes", "run_id": run.id}) finally: fb_db.close() except Exception as fb_err: - logger.explore( - "Background execute: unable to mark run FAILED", - extra={ - "src": "translate_routes", - "run_id": run.id, - "error": str(fb_err), - }, - ) + logger.explore("Background execute: unable to mark run FAILED", + extra={"src": "translate_routes", "run_id": run.id, "error": str(fb_err)}) finally: if bg_db is not None: bg_db.close() - threading.Thread( - target=_background_execute, - daemon=True, - ).start() + asyncio.create_task(_background_execute()) return _run_to_response(run) except ValueError as e: raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(e)) diff --git a/backend/src/core/utils/dataset_mapper.py b/backend/src/core/utils/dataset_mapper.py index 85dd7b2f..cec91af6 100644 --- a/backend/src/core/utils/dataset_mapper.py +++ b/backend/src/core/utils/dataset_mapper.py @@ -35,7 +35,7 @@ class DatasetMapper: # @PARAM database_id (int) - ID базы данных в Superset. # @PARAM sql_query (Optional[str]) - Произвольный SQL-запрос (должен вернуть column_name + verbose_name). # @RETURN Dict[str, str] - Словарь column_name -> verbose_name. - def get_sqllab_mappings( + async def get_sqllab_mappings( self, client: Any, dataset_id: int, @@ -72,7 +72,7 @@ class DatasetMapper: # Выполняем запрос через SQL Lab app_logger.info("[get_sqllab_mappings] Executing SQL Lab query on database %d...", database_id) - result = sqllab_executor.execute_and_poll(sql_query, database_id) + result = await sqllab_executor.execute_and_poll(sql_query, database_id) rows = result.get("results") or result.get("data") or result.get("result", []) app_logger.info("[get_sqllab_mappings] SQL Lab returned %d rows", len(rows)) @@ -123,7 +123,7 @@ class DatasetMapper: # @PARAM database_id (Optional[int]) - ID базы данных Superset (для sqllab source). # @PARAM sql_query (Optional[str]) - Произвольный SQL-запрос (для sqllab source). # @PARAM excel_path (Optional[str]) - Путь к XLSX файлу. - def run_mapping( + async def run_mapping( self, superset_client: Any, dataset_id: int, @@ -140,7 +140,7 @@ class DatasetMapper: try: if source == "sqllab": assert sqllab_executor and database_id, "sqllab_executor and database_id are required for sqllab source." - mappings.update(self.get_sqllab_mappings(superset_client, dataset_id, sqllab_executor, database_id, sql_query)) + mappings.update(await self.get_sqllab_mappings(superset_client, dataset_id, sqllab_executor, database_id, sql_query)) elif source == "excel": assert excel_path, "excel_path is required." mappings.update(self.load_excel_mappings(excel_path)) @@ -148,7 +148,7 @@ class DatasetMapper: app_logger.error("[run_mapping][Failure] Invalid source: %s.", source) return - dataset_response = superset_client.get_dataset(dataset_id) + dataset_response = await superset_client.get_dataset(dataset_id) dataset_data = dataset_response['result'] original_columns = dataset_data.get('columns', []) @@ -226,7 +226,7 @@ class DatasetMapper: } payload_for_update = {k: v for k, v in payload_for_update.items() if v is not None} - superset_client.update_dataset(dataset_id, payload_for_update) + await superset_client.update_dataset(dataset_id, payload_for_update) app_logger.info("[run_mapping][Success] Dataset %d columns' verbose_name updated.", dataset_id) else: app_logger.info("[run_mapping][State] No changes in columns' verbose_name, skipping update.") diff --git a/backend/src/plugins/mapper.py b/backend/src/plugins/mapper.py index c9a39954..3adb4450 100644 --- a/backend/src/plugins/mapper.py +++ b/backend/src/plugins/mapper.py @@ -169,7 +169,7 @@ class MapperPlugin(PluginBase): raise ValueError("database_id is required for sqllab source.") await executor.resolve_database_id(target_database_id=str(database_id)) - mapper.run_mapping( + await mapper.run_mapping( superset_client=client, dataset_id=dataset_id, source="sqllab", @@ -179,7 +179,7 @@ class MapperPlugin(PluginBase): ) else: # Excel source - mapper.run_mapping( + await mapper.run_mapping( superset_client=client, dataset_id=dataset_id, source="excel", diff --git a/backend/src/plugins/translate/_batch_insert.py b/backend/src/plugins/translate/_batch_insert.py index 35b6fa3f..09afb989 100644 --- a/backend/src/plugins/translate/_batch_insert.py +++ b/backend/src/plugins/translate/_batch_insert.py @@ -25,7 +25,7 @@ from .superset_executor import SupersetSqlLabExecutor # #region insert_batch_to_target [C:3] [TYPE Function] [SEMANTICS translate, insert, orchestrate] # @BRIEF Insert successful records from a single batch into the target table. -def insert_batch_to_target( +async def insert_batch_to_target( db: Session, config_manager: ConfigManager, job: TranslationJob, batch_id: str, run_id: str, ) -> None: @@ -45,7 +45,7 @@ def insert_batch_to_target( columns = [effective_target or "translated_text"] rows_for_sql = [{columns[0]: rec.target_sql or ""} for rec in records] - dialect, executor = _resolve_insert_backend(config_manager, job, batch_id) + dialect, executor = await _resolve_insert_backend(config_manager, job, batch_id) if dialect is None: return @@ -64,7 +64,7 @@ def insert_batch_to_target( for sql, chunk_count in statements: if sql is None: continue - _execute_insert_sql(executor, sql, batch_id, chunk_count) + await _execute_insert_sql(executor, sql, batch_id, chunk_count) total_inserted += chunk_count logger.reason(f"Batch {batch_id[:12]} inserted {total_inserted} rows", {"batch_id": batch_id, "rows": total_inserted, "chunks": len(statements)}) @@ -193,14 +193,14 @@ def _build_insert_rows( # #region _resolve_insert_backend [C:1] [TYPE Function] -def _resolve_insert_backend( +async def _resolve_insert_backend( config_manager: ConfigManager, job: TranslationJob, batch_id: str, ) -> tuple[str | None, SupersetSqlLabExecutor | None]: """Resolve the database backend and SQL executor for batch insert.""" try: env_id = job.environment_id or job.source_dialect or "" executor = SupersetSqlLabExecutor(config_manager, env_id) - executor.resolve_database_id(target_database_id=job.target_database_id) + await 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 for batch insert", @@ -233,14 +233,14 @@ def _generate_insert_sql( # #region _execute_insert_sql [C:1] [TYPE Function] -def _execute_insert_sql(executor: SupersetSqlLabExecutor, sql: str, batch_id: str, row_count: int) -> None: +async def _execute_insert_sql(executor: SupersetSqlLabExecutor, sql: str, batch_id: str, row_count: int) -> None: """Execute the INSERT SQL via Superset SQL Lab.""" try: - result = executor.execute_and_poll(sql=sql, max_polls=30, poll_interval_seconds=2.0) + result = await executor.execute_and_poll(sql=sql, max_polls=30, poll_interval_seconds=2.0) except Exception as e: logger.explore("Superset SQL submission failed for batch", {"batch_id": batch_id, "error": str(e)}) return - logger.reason(f"Batch {batch_id[:12]} inserted {row_count} rows", + logger.reason(f"Chunk inserted for batch {batch_id[:12]}", {"batch_id": batch_id, "rows": row_count, "status": result.get("status")}) # #endregion _execute_insert_sql # #endregion BatchInsertService diff --git a/backend/src/plugins/translate/_batch_proc.py b/backend/src/plugins/translate/_batch_proc.py index c92dd2f0..19011137 100644 --- a/backend/src/plugins/translate/_batch_proc.py +++ b/backend/src/plugins/translate/_batch_proc.py @@ -253,7 +253,7 @@ class BatchProcessingService: return result # -- Batch insert (delegation) -- - def insert_batch_to_target(self, job: TranslationJob, batch_id: str, run_id: str) -> None: - insert_batch_to_target(self.db, self.config_manager, job, batch_id, run_id) + async def insert_batch_to_target(self, job: TranslationJob, batch_id: str, run_id: str) -> None: + await insert_batch_to_target(self.db, self.config_manager, job, batch_id, run_id) # #endregion BatchProcessingService # #endregion BatchProcessingService diff --git a/backend/src/plugins/translate/executor.py b/backend/src/plugins/translate/executor.py index fde4fa48..611ec612 100644 --- a/backend/src/plugins/translate/executor.py +++ b/backend/src/plugins/translate/executor.py @@ -148,7 +148,7 @@ class TranslationExecutor: batch_id = batch_result.get("batch_id") if batch_id and batch_result.get("successful", 0) > 0: try: - self._insert_batch_to_target(job, batch_id, run.id) + await self._insert_batch_to_target(job, batch_id, run.id) except Exception as e: logger.explore("Batch INSERT failed (non-fatal)", {"batch_id": batch_id, "error": str(e)}) @@ -247,9 +247,9 @@ class TranslationExecutor: job, run_id, batch_index, batch_rows, dict_snapshot_hash, config_hash, preview_edits_cache=self._preview_edits_cache) - def _insert_batch_to_target(self, job, batch_id, run_id) -> None: + async def _insert_batch_to_target(self, job, batch_id, run_id) -> None: from ._batch_proc import BatchProcessingService - BatchProcessingService(self.db, self.config_manager).insert_batch_to_target(job, batch_id, run_id) + await BatchProcessingService(self.db, self.config_manager).insert_batch_to_target(job, batch_id, run_id) async def _call_llm_for_batch(self, job, run_id, batch_rows, dict_matches, batch_id, max_tokens=8192, _recursion_depth=0) -> dict: from ._llm_call import LLMTranslationService diff --git a/backend/src/plugins/translate/orchestrator.py b/backend/src/plugins/translate/orchestrator.py index 22b17c0a..39e08693 100644 --- a/backend/src/plugins/translate/orchestrator.py +++ b/backend/src/plugins/translate/orchestrator.py @@ -84,14 +84,12 @@ class TranslationOrchestrator: # @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( + async def execute_run( self, run: TranslationRun, - on_batch_progress: Callable[[str, int, int, int, int], None] | None = None, - skip_insert: bool = False, ) -> TranslationRun: with belief_scope("TranslationOrchestrator.execute_run"): - return self._runner.execute_run( + return await self._runner.execute_run( run=run, on_batch_progress=on_batch_progress, skip_insert=skip_insert, @@ -119,7 +117,7 @@ class TranslationOrchestrator: # region _generate_and_insert_sql [TYPE Function] [SEMANTICS backward-compat wrapper] # @PURPOSE: Backward-compatible delegating wrapper for SQL generation and insert. # @SIDE_EFFECT Delegates to SQLInsertService. May call Superset API. - def _generate_and_insert_sql( + async def _generate_and_insert_sql( self, job: Any, run: TranslationRun, @@ -127,7 +125,7 @@ class TranslationOrchestrator: """Backward-compatible wrapper — delegates to SQLInsertService.""" from .orchestrator_sql import SQLInsertService svc = SQLInsertService(self.db, self.config_manager, self.event_log) - return svc.generate_and_insert_sql(job, run) + return await svc.generate_and_insert_sql(job, run) # endregion _generate_and_insert_sql # region _update_language_stats [TYPE Function] [SEMANTICS backward-compat wrapper] diff --git a/backend/src/plugins/translate/orchestrator_exec.py b/backend/src/plugins/translate/orchestrator_exec.py index 0f896409..54209cc7 100644 --- a/backend/src/plugins/translate/orchestrator_exec.py +++ b/backend/src/plugins/translate/orchestrator_exec.py @@ -55,7 +55,7 @@ class TranslationExecutionEngine: # @PRE run is in PENDING status. # @POST Run executed, SQL generated, Superset submission attempted. # @SIDE_EFFECT LLM calls, DB writes, Superset API calls. - def execute_run( + async def execute_run( self, run: TranslationRun, on_batch_progress: Callable[[str, int, int, int, int], None] | None = None, @@ -89,7 +89,7 @@ class TranslationExecutionEngine: if run.status == "CANCELLED": return _complete_cancelled(self.db, self.event_log, self._aggregator, run, language_stats_map, self.current_user) - return _complete_success( + return await _complete_success( self.db, self.event_log, self._aggregator, self._sql_service, run, job, skip_insert, language_stats_map, self.current_user, ) diff --git a/backend/src/plugins/translate/orchestrator_run_completion.py b/backend/src/plugins/translate/orchestrator_run_completion.py index 5c52e6af..6eadd7ba 100644 --- a/backend/src/plugins/translate/orchestrator_run_completion.py +++ b/backend/src/plugins/translate/orchestrator_run_completion.py @@ -69,7 +69,7 @@ def complete_cancelled( # region complete_success [TYPE Function] # @PURPOSE: Finalize a successful run — update stats, optionally insert SQL, commit. # @SIDE_EFFECT DB writes; event log; Superset API call if skip_insert is False. -def complete_success( +async def complete_success( db, event_log, aggregator, @@ -102,7 +102,7 @@ def complete_success( db.commit() return run - insert_result = sql_service.generate_and_insert_sql(job, run) + insert_result = await sql_service.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 diff --git a/backend/src/plugins/translate/orchestrator_runner.py b/backend/src/plugins/translate/orchestrator_runner.py index 6a638771..f7fd2d28 100644 --- a/backend/src/plugins/translate/orchestrator_runner.py +++ b/backend/src/plugins/translate/orchestrator_runner.py @@ -41,14 +41,14 @@ class TranslationStageRunner: # @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( + async def execute_run( self, run: TranslationRun, on_batch_progress: Callable[[str, int, int, int, int], None] | None = None, skip_insert: bool = False, ) -> TranslationRun: with belief_scope("TranslationStageRunner.execute_run"): - return self._executor_engine.execute_run( + return await self._executor_engine.execute_run( run=run, on_batch_progress=on_batch_progress, skip_insert=skip_insert, diff --git a/backend/src/plugins/translate/orchestrator_sql.py b/backend/src/plugins/translate/orchestrator_sql.py index 3cbfc20d..366ec152 100644 --- a/backend/src/plugins/translate/orchestrator_sql.py +++ b/backend/src/plugins/translate/orchestrator_sql.py @@ -32,7 +32,7 @@ class SQLInsertService: # region generate_and_insert_sql [TYPE Function] # @PURPOSE: Generate INSERT SQL and submit to Superset SQL Lab. - def generate_and_insert_sql( + async def generate_and_insert_sql( self, job: TranslationJob, run: TranslationRun, @@ -64,7 +64,7 @@ class SQLInsertService: columns = [effective_target or "translated_text"] rows_for_sql = [{columns[0]: rec.target_sql or ""} for rec in records] - dialect = self._resolve_dialect(job) + dialect = await self._resolve_dialect(job) try: sql, row_count = SQLGenerator.generate( dialect=dialect, @@ -88,8 +88,8 @@ class SQLInsertService: 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) - result = executor.execute_and_poll(sql=sql, max_polls=30, poll_interval_seconds=2.0) + await executor.resolve_database_id(target_database_id=job.target_database_id) + result = await 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} @@ -103,11 +103,11 @@ class SQLInsertService: # endregion generate_and_insert_sql # region _resolve_dialect [TYPE Function] - def _resolve_dialect(self, job: TranslationJob) -> str: + async def _resolve_dialect(self, job: TranslationJob) -> str: 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) + await 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 dialect", {"error": str(e)})