busya
316d97ae25
- Convert all 84 contracts from legacy [DEF:] to #region/#endregion syntax - Fix complexity tiers: 14 modules re-tiered (6 C4 route modules, 7 C4→C5 plugin services) - Remove forbidden tags: @RATIONALE/@REJECTED stripped from C1–C4 contracts - Add required tags: @PRE/@POST/@SIDE_EFFECT on C4, @RELATION on C3, @DATA_CONTRACT/@INVARIANT on C5 - Add belief runtime markers (reason/reflect/explore) to 7 service.py functions - Fix @LAYER: route files → UI, plugins → Domain, superset_executor → Infra - Fix pre-existing test mock_service fixture in test_orchestrator.py - 196/196 translation tests pass, zero regressions
280 lines
13 KiB
Python
280 lines
13 KiB
Python
# #region TranslateRunRoutesModule [C:4] [TYPE Module] [SEMANTICS api,routes,translate,runs]
|
|
# @BRIEF Translation Run execution, history, status, records and batches routes.
|
|
# @LAYER UI
|
|
# @RELATION DEPENDS_ON -> [TranslationOrchestrator]
|
|
# @RELATION DEPENDS_ON -> [ConfigManager]
|
|
# @RELATION DEPENDS_ON -> [get_current_user]
|
|
# @RELATION DEPENDS_ON -> [get_db]
|
|
# @PRE ConfigManager and DB session initialized. User authenticated with translate permissions.
|
|
# @POST Translation run executed, manipulated, or queried. Results returned to caller.
|
|
# @SIDE_EFFECT Creates/updates TranslationRun records in DB; spawns background threads for execution; queries Superset API.
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, status, Query
|
|
from typing import Any, Dict, List, Optional
|
|
from sqlalchemy.orm import Session
|
|
|
|
from ....core.database import get_db, SessionLocal
|
|
from ....core.logger import logger, belief_scope
|
|
from ....schemas.auth import User
|
|
from ....dependencies import get_current_user, has_permission, get_config_manager
|
|
from ....core.config_manager import ConfigManager
|
|
from ....plugins.translate.orchestrator import TranslationOrchestrator
|
|
from ....plugins.translate.events import TranslationEventLog
|
|
from ....models.translate import TranslationRun
|
|
|
|
from ._router import router
|
|
from ._helpers import _run_to_response
|
|
|
|
|
|
# ============================================================
|
|
# Translation Run / Execute
|
|
# ============================================================
|
|
|
|
# #region run_translation [C:4] [TYPE Function] [SEMANTICS api,translate,runs]
|
|
# @BRIEF Execute a translation job (trigger a run).
|
|
# @PRE User has translate.job.execute permission. Job exists and preview is accepted.
|
|
# @POST Translation run created and started in background. Run object returned.
|
|
# @SIDE_EFFECT Creates TranslationRun record in DB; spawns background thread for translation execution.
|
|
# @RELATION DEPENDS_ON -> [TranslationOrchestrator]
|
|
@router.post("/jobs/{job_id}/run", status_code=status.HTTP_201_CREATED)
|
|
async def run_translation(
|
|
job_id: str,
|
|
full_translation: bool = Query(False, description="If True, fetch ALL rows from Superset dataset instead of preview-only rows"),
|
|
current_user: User = Depends(get_current_user),
|
|
_ = Depends(has_permission("translate.job", "EXECUTE")),
|
|
db: Session = Depends(get_db),
|
|
config_manager: ConfigManager = Depends(get_config_manager),
|
|
):
|
|
"""Execute a translation job (trigger a run).
|
|
|
|
By default runs translation on preview-approved rows only.
|
|
Set full_translation=true to fetch ALL rows from the Superset source dataset.
|
|
"""
|
|
logger.info(f"[translate_routes][run_translation] Job: {job_id}, User: {current_user.username}, full={full_translation}")
|
|
try:
|
|
orch = TranslationOrchestrator(db, config_manager, current_user.username)
|
|
run = orch.start_run(job_id=job_id, is_scheduled=False)
|
|
# Execute asynchronously in background with a FRESH session.
|
|
# The request-scoped session from Depends(get_db) is closed after the handler returns,
|
|
# so the background thread must create its own session to avoid "This transaction is closed".
|
|
import threading
|
|
|
|
def _execute_background():
|
|
thread_db = SessionLocal()
|
|
try:
|
|
thread_orch = TranslationOrchestrator(thread_db, config_manager, current_user.username)
|
|
# Reload run from DB with the new session (the passed run object is detached)
|
|
thread_run = thread_db.query(TranslationRun).filter(TranslationRun.id == run.id).first()
|
|
if thread_run:
|
|
thread_orch.execute_run(thread_run, full_translation=full_translation)
|
|
except Exception as e:
|
|
logger.error(f"[translate_routes][run_translation] Background execution error: {e}")
|
|
finally:
|
|
thread_db.close()
|
|
|
|
threading.Thread(target=_execute_background, daemon=True).start()
|
|
return _run_to_response(run)
|
|
except ValueError as e:
|
|
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(e))
|
|
except Exception as e:
|
|
logger.error(f"[translate_routes][run_translation] Error: {e}")
|
|
raise HTTPException(status_code=status.HTTP_502_BAD_GATEWAY, detail=f"Run failed: {e}")
|
|
# #endregion run_translation
|
|
|
|
|
|
# #region retry_run [C:4] [TYPE Function] [SEMANTICS api,translate,runs]
|
|
# @BRIEF Retry failed batches in a translation run.
|
|
# @PRE User has translate.job.execute permission. Run exists and has failed batches.
|
|
# @POST Failed batches re-executed. Updated run returned.
|
|
# @SIDE_EFFECT Updates TranslationRun and TranslationBatch records in DB.
|
|
# @RELATION DEPENDS_ON -> [TranslationOrchestrator]
|
|
@router.post("/runs/{run_id}/retry")
|
|
async def retry_run(
|
|
run_id: str,
|
|
current_user: User = Depends(get_current_user),
|
|
_ = Depends(has_permission("translate.job", "EXECUTE")),
|
|
db: Session = Depends(get_db),
|
|
config_manager: ConfigManager = Depends(get_config_manager),
|
|
):
|
|
"""Retry failed batches in a translation run."""
|
|
logger.info(f"[translate_routes][retry_run] Run: {run_id}, User: {current_user.username}")
|
|
try:
|
|
orch = TranslationOrchestrator(db, config_manager, current_user.username)
|
|
run = orch.retry_failed_batches(run_id)
|
|
return _run_to_response(run)
|
|
except ValueError as e:
|
|
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(e))
|
|
except Exception as e:
|
|
logger.error(f"[translate_routes][retry_run] Error: {e}")
|
|
raise HTTPException(status_code=status.HTTP_502_BAD_GATEWAY, detail=f"Retry failed: {e}")
|
|
# #endregion retry_run
|
|
|
|
|
|
# #region retry_insert [C:4] [TYPE Function] [SEMANTICS api,translate,runs]
|
|
# @BRIEF Retry the SQL insert phase for a completed run.
|
|
# @PRE User has translate.job.execute permission. Run completed with insert_status FAILED.
|
|
# @POST SQL insert phase re-executed. Updated run returned.
|
|
# @SIDE_EFFECT Re-executes SQL insert into Superset; updates TranslationRun insert_status.
|
|
# @RELATION DEPENDS_ON -> [TranslationOrchestrator]
|
|
@router.post("/runs/{run_id}/retry-insert")
|
|
async def retry_insert(
|
|
run_id: str,
|
|
current_user: User = Depends(get_current_user),
|
|
_ = Depends(has_permission("translate.job", "EXECUTE")),
|
|
db: Session = Depends(get_db),
|
|
config_manager: ConfigManager = Depends(get_config_manager),
|
|
):
|
|
"""Retry the SQL insert phase for a completed run."""
|
|
logger.info(f"[translate_routes][retry_insert] Run: {run_id}, User: {current_user.username}")
|
|
try:
|
|
orch = TranslationOrchestrator(db, config_manager, current_user.username)
|
|
run = orch.retry_insert(run_id)
|
|
return _run_to_response(run)
|
|
except ValueError as e:
|
|
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(e))
|
|
except Exception as e:
|
|
logger.error(f"[translate_routes][retry_insert] Error: {e}")
|
|
raise HTTPException(status_code=status.HTTP_502_BAD_GATEWAY, detail=f"Retry insert failed: {e}")
|
|
# #endregion retry_insert
|
|
|
|
|
|
# #region cancel_run [C:4] [TYPE Function] [SEMANTICS api,translate,runs]
|
|
# @BRIEF Cancel a running translation.
|
|
# @PRE User has translate.job.execute permission. Run is in RUNNING state.
|
|
# @POST Run is cancelled. Updated run returned.
|
|
# @SIDE_EFFECT Updates TranslationRun status to CANCELLED in DB.
|
|
# @RELATION DEPENDS_ON -> [TranslationOrchestrator]
|
|
@router.post("/runs/{run_id}/cancel")
|
|
async def cancel_run(
|
|
run_id: str,
|
|
current_user: User = Depends(get_current_user),
|
|
_ = Depends(has_permission("translate.job", "EXECUTE")),
|
|
db: Session = Depends(get_db),
|
|
config_manager: ConfigManager = Depends(get_config_manager),
|
|
):
|
|
"""Cancel a running translation."""
|
|
logger.info(f"[translate_routes][cancel_run] Run: {run_id}, User: {current_user.username}")
|
|
try:
|
|
orch = TranslationOrchestrator(db, config_manager, current_user.username)
|
|
run = orch.cancel_run(run_id)
|
|
return _run_to_response(run)
|
|
except ValueError as e:
|
|
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(e))
|
|
# #endregion cancel_run
|
|
|
|
|
|
# #region get_run_history [C:3] [TYPE Function] [SEMANTICS api,translate,runs]
|
|
# @BRIEF Get run history for a translation job.
|
|
# @RELATION DEPENDS_ON -> [TranslationOrchestrator]
|
|
@router.get("/jobs/{job_id}/runs")
|
|
async def get_run_history(
|
|
job_id: str,
|
|
page: int = Query(1, ge=1),
|
|
page_size: int = Query(20, ge=1, le=100),
|
|
current_user: User = Depends(get_current_user),
|
|
_ = Depends(has_permission("translate.history", "VIEW")),
|
|
db: Session = Depends(get_db),
|
|
config_manager: ConfigManager = Depends(get_config_manager),
|
|
):
|
|
"""Get run history for a translation job."""
|
|
logger.info(f"[translate_routes][get_run_history] Job: {job_id}, User: {current_user.username}")
|
|
try:
|
|
orch = TranslationOrchestrator(db, config_manager, current_user.username)
|
|
total, runs = orch.get_run_history(job_id, page=page, page_size=page_size)
|
|
return {"items": runs, "total": total, "page": page, "page_size": page_size}
|
|
except ValueError as e:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(e))
|
|
# #endregion get_run_history
|
|
|
|
|
|
# #region get_run_status [C:3] [TYPE Function] [SEMANTICS api,translate,runs]
|
|
# @BRIEF Get status and statistics for a translation run.
|
|
# @RELATION DEPENDS_ON -> [TranslationOrchestrator]
|
|
@router.get("/runs/{run_id}")
|
|
async def get_run_status(
|
|
run_id: str,
|
|
current_user: User = Depends(get_current_user),
|
|
_ = Depends(has_permission("translate.history", "VIEW")),
|
|
db: Session = Depends(get_db),
|
|
config_manager: ConfigManager = Depends(get_config_manager),
|
|
):
|
|
"""Get status and statistics for a translation run."""
|
|
logger.info(f"[translate_routes][get_run_status] Run: {run_id}, User: {current_user.username}")
|
|
try:
|
|
orch = TranslationOrchestrator(db, config_manager, current_user.username)
|
|
return orch.get_run_status(run_id)
|
|
except ValueError as e:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(e))
|
|
# #endregion get_run_status
|
|
|
|
|
|
# #region get_run_records [C:3] [TYPE Function] [SEMANTICS api,translate,runs]
|
|
# @BRIEF Get paginated records for a translation run.
|
|
# @RELATION DEPENDS_ON -> [TranslationOrchestrator]
|
|
@router.get("/runs/{run_id}/records")
|
|
async def get_run_records(
|
|
run_id: str,
|
|
page: int = Query(1, ge=1),
|
|
page_size: int = Query(50, ge=1, le=500),
|
|
status: Optional[str] = Query(None),
|
|
current_user: User = Depends(get_current_user),
|
|
_ = Depends(has_permission("translate.history", "VIEW")),
|
|
db: Session = Depends(get_db),
|
|
config_manager: ConfigManager = Depends(get_config_manager),
|
|
):
|
|
"""Get paginated records for a translation run."""
|
|
logger.info(f"[translate_routes][get_run_records] Run: {run_id}, User: {current_user.username}")
|
|
try:
|
|
orch = TranslationOrchestrator(db, config_manager, current_user.username)
|
|
return orch.get_run_records(run_id, page=page, page_size=page_size, status_filter=status)
|
|
except ValueError as e:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(e))
|
|
# #endregion get_run_records
|
|
|
|
|
|
# ============================================================
|
|
# Batches
|
|
# ============================================================
|
|
|
|
# #region get_batches [C:3] [TYPE Function] [SEMANTICS api,translate,runs]
|
|
# @BRIEF Get batches for a translation run.
|
|
# @RELATION DEPENDS_ON -> [TranslationBatch]
|
|
@router.get("/runs/{run_id}/batches")
|
|
async def get_batches(
|
|
run_id: str,
|
|
current_user: User = Depends(get_current_user),
|
|
_ = Depends(has_permission("translate.job", "VIEW")),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
"""Get batches for a translation run."""
|
|
logger.info(f"[translate_routes][get_batches] Run: {run_id}, User: {current_user.username}")
|
|
try:
|
|
from ....models.translate import TranslationBatch
|
|
batches = (
|
|
db.query(TranslationBatch)
|
|
.filter(TranslationBatch.run_id == run_id)
|
|
.order_by(TranslationBatch.batch_index.asc())
|
|
.all()
|
|
)
|
|
return [
|
|
{
|
|
"id": b.id,
|
|
"run_id": b.run_id,
|
|
"batch_index": b.batch_index,
|
|
"status": b.status,
|
|
"total_records": b.total_records or 0,
|
|
"successful_records": b.successful_records or 0,
|
|
"failed_records": b.failed_records or 0,
|
|
"started_at": b.started_at.isoformat() if b.started_at else None,
|
|
"completed_at": b.completed_at.isoformat() if b.completed_at else None,
|
|
"created_at": b.created_at.isoformat() if b.created_at else None,
|
|
}
|
|
for b in batches
|
|
]
|
|
except Exception as e:
|
|
logger.error(f"[translate_routes][get_batches] Error: {e}")
|
|
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(e))
|
|
# #endregion get_batches
|
|
|
|
# #endregion TranslateRunRoutesModule
|