actions

etl_craft.handlers.sql.actions

The eight SQL actions. Each wraps the task's SELECT in the writes it stands for.

• CREATE_TABLE replaces the target from the staged SELECT with rollback or recovery.
• SETUP_TABLE creates the target, empty, from the SELECT's shape plus the audit columns of the action that writes it, when it does not exist yet; an existing target is left alone.
• OVERWRITE_TABLE replaces the rows with atomic publication or row recovery.
• APPEND_TABLE inserts the SELECT's rows, without comparing the shapes.
• SCD1_MERGE updates changed rows in place by merge key and inserts new keys.
• SCD2_MERGE closes the active version of a changed key (ACTIVE_FLAG='N') and inserts a new one.

In both merges a key soft-deleted by DELETE_ROWS that the SELECT returns again comes back (DELETE_FLAG='N'), changed or not. A NULL in a merge key column fails the task. - DROP_TABLE drops the target if it exists, once this pipeline's CREATE_TABLE task for it has succeeded in the run. - DELETE_ROWS deletes, or flags DELETE_FLAG='Y', the target rows whose merge key the SELECT returns; rows already flagged are left as they are.

Only CREATE_TABLE and SETUP_TABLE create tables; every other action fails, naming the remedy, when its target does not exist.

Each reports the counts it can know: source rows, target rows after the write, and the rows inserted, updated or deleted.

A row counts as changed when its HASH_KEY, an MD5 over MERGE_COMPARE_COLUMNS, differs. Merge updates join the deduplicated stage using the warehouse's UPDATE FROM or matched MERGE strategy. SCD2 materializes changed keys before closing active versions so its insert phase can still find those keys afterwards.

ActionContext dataclass

What every action needs besides the warehouse session.

select_sql property

select_sql: str

The task's SELECT; every action but DROP_TABLE has one.

stamp property

stamp: dict[str, object]

The values the audit columns are stamped with.

create_table

create_table(session: Session, action: ActionContext) -> HandlerResult

Replace the target with the SELECT's rows, stamped with the run id.

setup_table

setup_table(session: Session, action: ActionContext) -> HandlerResult

Create the target, empty, when it does not exist; leave an existing one alone.

Its columns are the SELECT's, then PIPELINE_RUN_ID, the audit columns of the action that writes it and ROW_ID. That action is SETUP_FOR when set, else the one the pipeline's other tasks on the target write with.

writer_action

writer_action(setup_for: str | None, others: list[TargetTask], target: str) -> str

Return the writer whose audit columns a SETUP_TABLE target gets.

others are the pipeline's other tasks on the target.

overwrite_table

overwrite_table(session: Session, action: ActionContext) -> HandlerResult

Empty the target and insert the SELECT's rows.

append_table

append_table(session: Session, action: ActionContext) -> HandlerResult

Replace this task run's appended batch, retaining rows from other loads.

TASK_RUN_ID makes retries converge after an insert committed without its task outcome, so a target without it is refused before anything is staged, naming the command that adds it.

scd1_merge

scd1_merge(session: Session, action: ActionContext) -> HandlerResult

Update changed rows in place by merge key and insert new keys.

With PRESERVE_TARGET, a NULL in the SELECT keeps the target's value, and the hash is taken over the values actually kept.

scd2_merge

scd2_merge(session: Session, action: ActionContext) -> HandlerResult

Close the active version of each changed key and insert a new active version.

A key with no active version, new or left without one by an interrupted earlier run, gets one inserted, so a rerun converges.

drop_table

drop_table(session: Session, action: ActionContext) -> HandlerResult

Drop the target if it exists, only once this pipeline's CREATE_TABLE task ran this run.

A target already gone is not an error: the drop has nothing left to do.

delete_rows

delete_rows(session: Session, action: ActionContext) -> HandlerResult

Delete, or flag DELETE_FLAG='Y', the target rows whose merge key the SELECT returns.