Ingestion scripts¶

A task with HANDLER = 'PYTHON' runs one of your team's Python scripts to bring data in: it reads its source from where the last run left off and writes a warehouse table. The engine runs it, captures everything it prints and logs, records its counts, and keeps its offset.

The task¶

Parameter	Value
`SCRIPT_NAME`	the script, a path inside the project's `ingestion_scripts/` folder, such as `crm/customers.py`
`INPUT_PARAMS`	optional: a JSON object the script receives as a dictionary, such as `{"region": "eu", "days": 30}`
`TARGET_OBJECT`, `SOURCE_OBJECT`	optional: the table the script writes and where it reads from, for the catalog's lineage graphs

The script¶

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:
    region, days = task.input_params["region"], task.input_params["days"]
    since = task.offset.value if task.offset else 0  # None on the first run
    rows = fetch_orders(region, after_id=since, days=days)  # your own code
    task.logger.info("fetched %d orders for %s", len(rows), region)
    with task.warehouse() as engine, engine.begin() as conn:
        written = insert_orders(
            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,
    )

ScriptTask gives the script:

Attribute	Holds
`offset`	where the last successful run left off, an `Offset` with its `value` and `datatype`, or `None` on the first run
`input_params`	`INPUT_PARAMS` as a dictionary, empty when unset
`pipeline_id`	the pipeline definition's id; stamp it on every row you write
`pipeline_run_id`	the pipeline execution's id; stamp it on every row you write
`task_run_id`	this task execution's id, shared by its retries; stamp it on every row you write
`refresh_type`	the pipeline's `FULL` or `INCREMENTAL`
`force`	true when the task was run with `--force`
`run_date`	the date the run runs as of: the day it started (UTC), the `--run-date` it was given, or the date of a backfill run
`backfill`	true in a backfill run: `offset` is then `None`, and an offset the script returns is not stored, so read the source for `run_date`
`warehouse()`	a connection to the warehouse, queued behind other writers where the warehouse allows only one (a DuckDB file)
`table(name)`	`schema.table` as its full name in the active warehouse database, such as `analytics.raw.orders`, so the same script writes to each environment's database
`logger`	a logger named after the task
`task_params`, `config`	the task's parameters and the loaded `craft-connector.yml`

ScriptResult holds row_count, the rows the script wrote, a required whole number that the engine records as the task's source, target and insert counts; offset, the new offset, or None to keep the stored one; and variables, a mapping of names to any values of your own, listed in the task log. The engine validates the whole result before storing its offset.

A script that needs neither the offset nor INPUT_PARAMS can define run() without a parameter; it still returns a ScriptResult.

An offset is a value and its datatype, and the datatype is always declared: Offset(1042, "NUMBER"), Offset("cursor-9", "TEXT") or Offset(latest, "TIMESTAMP"), or the shorthands Offset.number(1042), Offset.text("cursor-9") and Offset.timestamp(latest). The value is cast to its datatype, as CAST(value AS type) would in SQL: Offset("1042", "NUMBER") holds the number 1042, and Offset("abc", "NUMBER") fails. The type is never guessed from the value.

The offset is stored as text beside its datatype, so one column holds every kind. The next run gets it back cast to that datatype: task.offset.value is an int or Decimal for NUMBER, a str for TEXT and a datetime for TIMESTAMP, and task.offset.datatype names it. Timestamp offsets keep microsecond precision. A datetime with nonzero nanoseconds, such as a pandas Timestamp, is refused; round it to microseconds before building the offset.

The offset is stored in AUD_TASK_OFFSET_TRACKER if and only if the attempt is recorded SUCCESS. The attempt outcome, task summary, new offset and admitted task-dependency consumption commit in one Engine DB transaction. A crash before that commit preserves the previous offset; a crash after it preserves the complete success. A failed attempt is retried from the same place, and the offset keeps its datatype from one run to the next. Warehouse writes made by the script commit separately; make them safe to repeat when an attempt must be retried.

Scripts may import helper modules kept beside them in ingestion_scripts/. Each script is compiled from its current source and registered as a module named etl_craft_script_ followed by its path relative to ingestion_scripts/, with / and . replaced by _: crm/customers.py becomes etl_craft_script_crm_customers_py. The script's own future imports apply; dataclasses, enums, pickling and typing.get_type_hints work as in other modules. A script may use ProcessPoolExecutor with the fork start method on platforms that support it. The spawn start method cannot re-import the script by its generated module name; put functions needed by spawned workers in a normally importable helper module.

Output and logs¶

The script runs inside the task's own process. Everything it prints, writes to stderr or logs through logging goes to the attempt's log file, and log records carry the task's context. HTTP and SDK libraries (urllib3, httpx, httpcore, requests, botocore, boto3 and s3transfer) are captured at WARNING or above unless etl-craft's log level is DEBUG, to keep request URLs and credential query strings out of routine logs. DEBUG can include those details. A script must never print or log secrets, including in its own request URLs or exception messages.

For example:

2026-09-25 10:15:02,114 INFO etl_craft_script.load_customers [task_run_id=412 pipeline=CRM task=load_customers pipeline_run_id=97 attempt=1]: fetched 120 orders for eu

The engine adds its own lines around it: the offset and inputs the script started from, and what it returned. AUD_TASK_RUN_LOG records the row count as the source, target and insert counts, and TASK_LOG lists them with the new offset and the script's variables:

SOURCE_COUNT = 120
TARGET_COUNT = 120
INSERT_COUNT = 120
OFFSET = 1042 (NUMBER)

When something is wrong¶

The task fails with a message that names the script and the problem when:

SCRIPT_NAME is missing, points outside ingestion_scripts/, or names no .py file there;
INPUT_PARAMS is not a JSON object;
the script cannot be imported, or defines no run, or a run that takes more than the task;
run raises: the message has the exception, and the traceback is in the attempt's log;
run calls sys.exit: raise an exception to fail, return a ScriptResult to succeed;
run returns something other than a ScriptResult, a row_count that is not a whole number of 0 or more, an offset of a different type from the stored one, or variables that is not a mapping;
a timestamp offset has nanoseconds the offset store cannot keep.

The task's time limit covers the script: a script still running when it passes is stopped, and the task is recorded FAILED.