transitions
etl_craft.engine.transitions ¶
Guarded lifecycle writes and atomic task-attempt summaries.
ActiveAttempt
dataclass
¶
Identity and owner of a nonterminal attempt.
create_active_run ¶
create_active_run(conn: Connection, pipeline_id: int, *, run_date: date | None = None, backfill: bool = False, trigger_kind: Literal['MANUAL', 'BACKFILL', 'STAND_IN'] | None = None) -> int | None
Start a new IN-PROGRESS run of pipeline_id and return its id.
Return None when the pipeline already has a run in progress: the unique index on
IN-PROGRESS runs refuses a second one, so of several processes starting a run at once,
exactly one gets an id. A new run runs as of run_date (today, in UTC, unless given),
and backfill marks it part of a backfill.
end_run_if ¶
End pipeline_run_id with status only if it is still from_status.
Return whether it did. The caller decides what a refusal means; nothing is changed then.
resolve_run ¶
resolve_run(conn: Connection, pipeline_id: int, *, force: bool = False, orchestrated: bool = False, reason: str = 'task requested under an ended run', selector: RunSelector = runlog.ACTIVE_RUN) -> tuple[int, str | None]
Bind to the exact selected run; force or an orchestrator may reopen it.
find_or_create_task_run ¶
find_or_create_task_run(conn: Connection, task_id: int, pipeline_run_id: int) -> runlog.TaskRunBinding
Return the row of task_id under pipeline_run_id, creating it IN-PROGRESS.
A concurrent insert that loses reads back the winner.
finish_task_run ¶
finish_task_run(conn: Connection, task_run_id: int, *, status: str, counts: Counts = NO_COUNTS, error_message: str | None = None, task_log: str | None = None, attempt_id: int | None = None, owner: str | None = None, offset: StoredOffset | None = None, retryable: bool = True) -> None
Record the current attempt's outcome on its row; each attempt writes only its own counts.
finalize_pipeline_run ¶
finalize_pipeline_run(conn: Connection, pipeline_run_id: int, status: str, *, sla_in_hours: float | None = None, owner: str | None = None, consume: bool = False) -> runlog.RunEnding
End pipeline_run_id with status and END_DATE now, if it is still IN-PROGRESS.
With sla_in_hours (the pipeline's SLA_IN_HOURS, whenever it has one) the run is also
marked MET or BREACHED, measured from START_DATE, whether or not SLA emails are on,
unless it already has an SLA status: a run reopened after it met its SLA stays MET, and
the returned result carries the recorded status. STATUS is left alone either way: a late run
did its work. A run that is no longer IN-PROGRESS is not changed, and ended is false.
mark_sla_breached ¶
Mark a still running pipeline_run_id BREACHED; return whether this call did.
False when the run already finished or was already marked.
mark_task_run ¶
mark_task_run(conn: Connection, task_run_id: int, *, status: str, error_message: str, target_count: int | None) -> None
Set a task row's status as an operator marked it.
cancel_task_run ¶
End an IN-PROGRESS task row CANCELLED; return whether it was still running.
cancel_pipeline_run ¶
Cancel an active run on request; refuse a changed status or owner.
delete_skipped_task_run ¶
Remove a SKIPPED row of a task that never ran, so the resumed run decides again.
create_run ¶
create_run(conn: Connection, pipeline_id: int, actor: Actor, *, run_date: date | None = None, trigger_kind: str = 'MANUAL', run_key: str | None = None, status: str = 'IN-PROGRESS') -> int
Create an active run or a queued/skipped schedule with its exact key.
admit_run ¶
Admit one queued run only while its pipeline has no active run.
start_run ¶
start_run(conn: Connection, pipeline_run_id: int, actor: Actor, *, owner: str, lease_expires_at: datetime) -> None
Attach the overseer to an active run that has no owner.
finish_run ¶
finish_run(conn: Connection, pipeline_run_id: int, status: str, actor: Actor, *, owner: str | None = None, sla_status: str | None = None) -> None
End an active run only when no claimed or running attempt remains.
reopen_run ¶
reopen_run(conn: Connection, pipeline_run_id: int, actor: Actor, *, owner: str | None = None, reason: str = 'run reopened') -> None
Reopen a terminal run and record REOPEN in the same transaction.
mark_run ¶
mark_run(conn: Connection, pipeline_run_id: int, status: str, actor: Actor, *, owner: str | None = None) -> None
Record an operator's outcome only when no claimed or running attempt remains.
queue_attempt ¶
queue_attempt(conn: Connection, task_run_id: int, actor: Actor, *, log_path: str | None = None, not_before: datetime | None = None, previous_attempt_id: int | None = None) -> int
Queue one attempt, serializing admission through its task summary.
claim_attempt ¶
claim_attempt(conn: Connection, attempt_id: int, actor: Actor, *, owner: str, lease_expires_at: datetime, host: str | None = None) -> None
Claim a queued attempt for one owner.
start_attempt ¶
start_attempt(conn: Connection, attempt_id: int, actor: Actor, *, owner: str, host: str | None = None, pid: int | None = None, process_start: str | None = None) -> None
Start a claimed attempt, fenced by its owner.
finish_attempt ¶
finish_attempt(conn: Connection, attempt_id: int, status: str, actor: Actor, *, owner: str, counts: Counts = NO_COUNTS, error_message: str | None = None, task_log: str | None = None, exit_code: int | None = None, offset: StoredOffset | None = None, retryable: bool = True) -> None
Record the owned outcome, summary, offset and admitted consumption atomically.
time_out_attempt ¶
time_out_attempt(conn: Connection, attempt_id: int, actor: Actor, *, owner: str, error_message: str, task_log: str | None = None) -> None
End an owned running attempt that exceeded its time limit.
cancel_attempt ¶
cancel_attempt(conn: Connection, attempt_id: int, actor: Actor, *, error_message: str, task_log: str | None = None) -> None
Cancel a nonterminal attempt on behalf of an operator or overseer.
lose_attempt ¶
lose_attempt(conn: Connection, attempt_id: int, actor: Actor, *, owner: str | None, error_message: str | None = None) -> None
Record an uncertain outcome for an owner whose lease expired.
renew_lease ¶
renew_lease(conn: Connection, attempt_id: int, actor: Actor, *, owner: str, lease_expires_at: datetime) -> None
Renew an owned live lease; an expired lease cannot be revived.
active_attempt ¶
Return the task's nonterminal attempt, if one exists.
restart_rule_run ¶
Restart a business-rule summary that has not succeeded.
finish_rule_run ¶
Record a business-rule summary's outcome.
assert_attempt ¶
Reject a result referring to another task, owner or superseded attempt.
set_task_log ¶
set_task_log(conn: Connection, task_run_id: int, attempt_id: int, owner: str, task_log: str | None) -> None
Append captured process output to the summary of this exact attempt.
ensure_attempt_started ¶
ensure_attempt_started(conn: Connection, attempt_id: int, actor: Actor, *, owner: str, host: str, pid: int, completed: bool = False, process_start: str | None = None) -> None
Accept the same start acknowledged by parent and child, with no second write.
create_rule_run ¶
Create a business-rule summary under its task run.
renew_run_lease ¶
renew_run_lease(conn: Connection, run_id: int, actor: Actor, *, owner: str, lease_expires_at: datetime) -> None
Renew a supervisor's run lease only while its existing lease is live.
release_run_lease ¶
Release this exact supervisor when no active attempt remains, optionally after expiry.