Skip to content

interventions

etl_craft.execution.interventions

mark and cancel: an operator's control over runs, in local mode.

In local mode etl-craft is the orchestrator, so it is where an operator steps in:

  • mark sets a task of the selected run, or the run itself, to SUCCESS, FAILED or SKIPPED, with a reason. A marked SUCCESS satisfies a HAS_DATA dependency only with a stated row count. Marking a task of a run that has ended reopens the run, and the tasks the engine skipped without running are reset, so running the pipeline again resumes it from there: mark a failed task SUCCESS and its dependents run.
  • mark --new-run records a finished stand-in run, with a task in it if named, so a gate on this pipeline passes where it cannot really run (an upstream that only exists elsewhere).
  • cancel ends the pipeline's run CANCELLED, with its running tasks. The process running each task watches for this and stops it; the process running the pipeline starts nothing more.
  • run --skip records a run SKIPPED on purpose, as mark --new-run does, so the pipelines that depend on it see a run that did nothing.
  • pause stops a pipeline from running until resume: run starts nothing of it, and a run in progress starts no more tasks and stays IN-PROGRESS, to go on once resumed. Pauses are recorded in AUD_PIPELINE_PAUSES.

Every change is recorded in AUD_RUN_INTERVENTIONS with what the row held before, who asked and why, and no attempt, log or error is erased. In remote mode the orchestrator is the only source of truth, so these are refused: mark or clear the task in the orchestrator instead.

Intervened dataclass

What an intervention did, in one line, and the run it changed.

mark_task 

mark_task(engine: Engine, config: ConnectorConfig, pipeline_code: str, task_code: str, status: str, reason: str, *, rows: int | None = None, stale: bool = False, requested_by: str | None = None, selector: RunSelector = runlog.ACTIVE_RUN) -> Intervened

Mark task_code status under the selected run.

A run that has ended is reopened, and the tasks the engine skipped without running are reset, so the next run --pipeline_code resumes it. A task that is IN-PROGRESS is refused while its lease is live. stale reconciles expired attempts before marking. Raises RunStateError when the pipeline has no run, and RunRefusedError in remote mode or while the task is running.

mark_run 

mark_run(engine: Engine, config: ConnectorConfig, pipeline_code: str, status: str, reason: str, *, requested_by: str | None = None, selector: RunSelector = runlog.ACTIVE_RUN) -> Intervened

Mark the selected run status, ending it now if it had not ended.

A downstream pipeline's gate judges the marked status. Its tasks keep theirs, and no upstream run is consumed. Raises RunRefusedError while a task of the run is running.

record_stand_in_run 

record_stand_in_run(engine: Engine, config: ConnectorConfig, pipeline_code: str, status: str, reason: str, *, task_code: str | None = None, rows: int | None = None, requested_by: str | None = None) -> Intervened

Record a finished run of pipeline_code that did not really run: mark --new-run.

Downstream gates judge it like any run: on the pipeline, or with task_code on that task, which gets status and rows. Raises RunStateError while the pipeline has a run in progress.

skip_run 

skip_run(engine: Engine, config: ConnectorConfig, pipeline_code: str, reason: str, *, requested_by: str | None = None) -> Intervened

Record a run of pipeline_code SKIPPED on purpose: run --skip.

Raises RunStateError while the pipeline has a run in progress.

pause_pipeline 

pause_pipeline(engine: Engine, config: ConnectorConfig, pipeline_code: str, reason: str, *, requested_by: str | None = None) -> str

Pause pipeline_code until it is resumed; return what was done, in one line.

Raises RunStateError when it is already paused.

resume_pipeline 

resume_pipeline(engine: Engine, config: ConnectorConfig, pipeline_code: str, reason: str, *, requested_by: str | None = None) -> str

Resume a paused pipeline_code; return what was done, in one line.

Raises RunStateError when it is not paused.

open_pause 

open_pause(engine: Engine, pipeline_code: str) -> pauses.Pause | None

Return the open pause of pipeline_code, or None when it is not paused.

cancel_run 

cancel_run(engine: Engine, config: ConnectorConfig, pipeline_code: str, reason: str, *, requested_by: str | None = None, selector: RunSelector = runlog.ACTIVE_RUN) -> Intervened

End the pipeline's queued or active run CANCELLED, with every task still running.

The process running each task stops it within a few seconds, and the process running the pipeline starts nothing more. Raises RunStateError when the selected run is not queued or active.

record_change 

record_change(engine: Engine, *, pipeline_id: int, pipeline_run_id: int, action: InterventionAction, reason: str, task_id: int | None = None, from_status: str | None = None, to_status: str | None = None) -> None

Record one change an operator made to a run, from its own transaction.

check_override 

check_override(config: ConnectorConfig, option: str, reason: str | None) -> str

Return the reason for option, an override of local mode's checks.

Raises RunRefusedError in remote mode and UsageError without a reason.

record_gate_bypass 

record_gate_bypass(engine: Engine, pipeline_id: int, pipeline_run_id: int, policy: GatePolicy, bypassed: Sequence[str], *, task_id: int | None = None) -> None

Record that Dependency_gates let a run, or a task of it, through unsatisfied gates.

reset_engine_skipped 

reset_engine_skipped(engine: Engine, pipeline_id: int, run_id: int, reason: str, *, only: Collection[int] | None = None) -> tuple[list[str], list[str]]

Reset, in a transaction of its own, the tasks the engine skipped without running.

only limits it to those task ids. Return the tasks reset and, one line each, those kept because a downstream consumed them.