121 lines
4.6 KiB
Python
121 lines
4.6 KiB
Python
# #region orchestrator_cancel [C:3] [TYPE Module] [SEMANTICS translate, cancel, retry_insert]
|
|
# @BRIEF Cancel and retry-insert operations for translation runs.
|
|
# @RELATION DEPENDS_ON -> [TranslationRun]
|
|
# @RELATION DEPENDS_ON -> [TranslationJob]
|
|
# @RELATION DEPENDS_ON -> [TranslationEventLog]
|
|
# @RELATION DEPENDS_ON -> [SQLInsertService]
|
|
# @RATIONALE Extracted from orchestrator_retry.py for INV_7 compliance (module < 150 lines).
|
|
|
|
from datetime import UTC, datetime
|
|
|
|
from sqlalchemy import text
|
|
from sqlalchemy.orm import Session
|
|
|
|
from ...core.config_manager import ConfigManager
|
|
from ...core.logger import belief_scope, logger
|
|
from ...models.translate import TranslationJob, TranslationRun
|
|
from .events import TranslationEventLog
|
|
from .orchestrator_sql import SQLInsertService
|
|
|
|
|
|
def _fallback_cancel_request(db: Session, run_id: str) -> TranslationRun:
|
|
"""Set cancellation flag when row lock prevents direct status update."""
|
|
try:
|
|
db.rollback()
|
|
except Exception:
|
|
pass
|
|
logger.explore("Row lock timeout — setting cancellation flag via direct SQL", {"run_id": run_id})
|
|
db.execute(
|
|
text("UPDATE translation_runs SET error_message = 'CANCEL_REQUESTED' WHERE id = :run_id"),
|
|
{"run_id": run_id},
|
|
)
|
|
db.commit()
|
|
run = db.query(TranslationRun).filter(TranslationRun.id == run_id).first()
|
|
if not run:
|
|
raise ValueError(f"Run '{run_id}' not found")
|
|
return run
|
|
|
|
|
|
# #region cancel_run [C:3] [TYPE Function] [SEMANTICS translate, cancel, run]
|
|
# @BRIEF Cancel a running translation run.
|
|
# @SIDE_EFFECT DB writes; event log.
|
|
# @RELATION DEPENDS_ON -> [TranslationRun]
|
|
# @RELATION DEPENDS_ON -> [TranslationEventLog]
|
|
def cancel_run(
|
|
db: Session,
|
|
event_log: TranslationEventLog,
|
|
current_user: str | None,
|
|
run_id: str,
|
|
) -> TranslationRun:
|
|
"""Cancel a translation run, handling lock timeout with direct SQL fallback."""
|
|
with belief_scope("orchestrator_cancel.cancel_run"):
|
|
try:
|
|
db.execute(text("SET LOCAL lock_timeout = '3s'"))
|
|
run = db.query(TranslationRun).filter(TranslationRun.id == run_id).first()
|
|
except Exception:
|
|
return _fallback_cancel_request(db, run_id)
|
|
|
|
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}'. "
|
|
"Only PENDING or RUNNING runs can be cancelled."
|
|
)
|
|
|
|
run.status = "CANCELLED"
|
|
run.completed_at = datetime.now(UTC)
|
|
try:
|
|
db.flush()
|
|
except Exception:
|
|
return _fallback_cancel_request(db, run_id)
|
|
event_log.log_event(
|
|
job_id=run.job_id, run_id=run.id, event_type="RUN_CANCELLED",
|
|
payload={}, created_by=current_user,
|
|
)
|
|
db.commit()
|
|
db.refresh(run)
|
|
return run
|
|
# #endregion cancel_run
|
|
|
|
|
|
# #region retry_insert [C:3] [TYPE Function] [SEMANTICS translate, retry, insert]
|
|
# @BRIEF Retry the SQL insert phase for a completed run.
|
|
# @SIDE_EFFECT Superset API call; DB writes.
|
|
# @RELATION DEPENDS_ON -> [TranslationRun]
|
|
# @RELATION DEPENDS_ON -> [TranslationJob]
|
|
# @RELATION DEPENDS_ON -> [TranslationEventLog]
|
|
# @RELATION DEPENDS_ON -> [SQLInsertService]
|
|
async def retry_insert(
|
|
db: Session,
|
|
config_manager: ConfigManager,
|
|
event_log: TranslationEventLog,
|
|
current_user: str | None,
|
|
run_id: str,
|
|
) -> TranslationRun:
|
|
"""Retry SQL insert for a completed run — generates and submits INSERT statements."""
|
|
with belief_scope("orchestrator_cancel.retry_insert"):
|
|
run = db.query(TranslationRun).filter(TranslationRun.id == run_id).first()
|
|
if not run:
|
|
raise ValueError(f"Run '{run_id}' not found")
|
|
job = db.query(TranslationJob).filter(TranslationJob.id == run.job_id).first()
|
|
if not job:
|
|
raise ValueError(f"Job '{run.job_id}' not found")
|
|
|
|
event_log.log_event(
|
|
job_id=job.id, run_id=run.id, event_type="RUN_RETRY_INSERT",
|
|
payload={}, created_by=current_user,
|
|
)
|
|
sql_service = SQLInsertService(db, config_manager, event_log)
|
|
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 "")
|
|
if insert_result.get("error_message"):
|
|
run.error_message = insert_result["error_message"]
|
|
db.flush()
|
|
db.commit()
|
|
db.refresh(run)
|
|
return run
|
|
# #endregion retry_insert
|
|
# #endregion orchestrator_cancel
|