scripting

etl_craft.scripting

What an ingestion script receives and returns: the HANDLER=PYTHON contract.

A task with HANDLER = 'PYTHON' names a script under the project's ingestion_scripts/ in SCRIPT_NAME. The script defines run, which takes a ScriptTask and returns a ScriptResult::

from etl_craft.scripting import Offset, ScriptResult, ScriptTask

def run(task: ScriptTask) -> ScriptResult:
    since = task.offset.value if task.offset else 0
    rows = read_source_after(since)            # the script's own code
    with task.warehouse() as engine, engine.begin() as conn:
        written = write(
            conn, rows, pipeline_id=task.pipeline_id,
            pipeline_run_id=task.pipeline_run_id, task_run_id=task.task_run_id,
        )
    return ScriptResult(
        row_count=written,
        offset=Offset.number(max(r.id for r in rows)) if rows else None,
    )

The script reads its source from where the last successful run left off (task.offset) and writes its table, stamping each row with task.pipeline_id, task.pipeline_run_id and task.task_run_id. It reports the rows it wrote, which the engine records as the task's source, target and insert counts, and, when it moved on, the new offset, which the engine stores for the next run. A script that needs neither the offset nor INPUT_PARAMS may define run() without a parameter. Everything it prints or logs goes to the task attempt's log.

Offset dataclass

Where a script left off: a value and its datatype, NUMBER, TEXT or TIMESTAMP.

The datatype is always given, never guessed from the value. The value is cast to it, as CAST(value AS datatype) would in SQL: Offset("7", "NUMBER") holds 7, and Offset("abc", "NUMBER") fails. A NUMBER is an int or a Decimal, a TEXT a str, a TIMESTAMP a datetime (ISO 8601 text is read as one). The offset is stored as text beside its datatype, and a script gets it back cast to that datatype. Offset.number, Offset.text and Offset.timestamp are shorthands.

__post_init__

__post_init__() -> None

Check the datatype and cast the value to it.

number classmethod

number(value: int | Decimal) -> Offset

Build a NUMBER offset, such as the largest id read.

text classmethod

text(value: str) -> Offset

Build a TEXT offset, such as a cursor or a file name.

timestamp classmethod

timestamp(value: datetime) -> Offset

Build a TIMESTAMP offset, such as the latest change read.

stored

stored() -> str

Return the text the offset is stored as, beside its datatype.

from_stored classmethod

from_stored(datatype: str, text: str) -> Offset

Read an offset back from its stored datatype and text, cast to that datatype.

ScriptResult dataclass

What a script reports: the rows it wrote, where it left off, and any values of its own.

row_count is required, a whole number of 0 or more. offset of None keeps the stored offset. variables are listed in the task log as NAME = value lines.

ScriptTask dataclass

What a script is given to run.

offset is where the last successful run left off, None on the first run. pipeline_id identifies the pipeline definition, pipeline_run_id its execution, and task_run_id this task's execution, shared by its retries. input_params is the task's INPUT_PARAMS, a JSON object, as a dictionary. force is true when the task was run with --force. run_date is the date the run runs as of: the day it started, or the date a backfill run is for. In a backfill (backfill), offset is None and an offset the script returns is not stored: a backfill reads its source for run_date, and leaves the next scheduled run where the last one left off.

table

table(name: str) -> str

Return schema.table as its full name in the active warehouse database.

database.schema.table is kept as written, so a script names its tables once and writes to the development, test or production database the profile selects.

warehouse

warehouse() -> Iterator[Engine]

Open the warehouse, queued behind other writers where it allows only one.