Execution¶

Assets execute in a DAG driven by a runner. The DAG orders operations by dependency; the runner schedules ready operations, tracks their state, and returns a result. The engine is async-native; every entry point has a sync form for scripts and notebooks.

Building a DAG¶

import interloper as il

dag = il.DAG(shop_source, finance_source, standalone_asset)

DAG(*items) accepts any workload: source or asset instances, definitions (instantiated for you), and jobs. Each is flattened into its operations. Construction validates the graph and raises on duplicates, missing or mismatched dependencies, cycles, and an unpartitioned asset downstream of a partitioned one (see Dependencies).

dag.operations                    # every node, in insertion order
dag.operation_map                 # id -> node
dag.topological_generations()     # lists of nodes that may run in parallel
dag.get_predecessors(asset.id)    # upstream ids
dag.get_successors(asset.id)      # downstream ids
dag.mini_dag(asset.id)            # one node plus read-only parents

Non-enabled nodes stay in the graph as dependencies but never execute and never appear in the generations.

Materializing¶

result = dag.materialize()                                   # default AsyncRunner
result = dag.materialize(il.TimePartition(dt.date(2026, 1, 15)))
result = await dag.materialize_async(window)

For a specific runner or its options:

runner = il.AsyncRunner(max_workers=8, fail_fast=False)
result = il.run(runner.run(dag, window, metadata={"run_id": "..."}))    # sync code
result = await runner.run(dag, window)                                  # async code

window is a PartitionWindow, or a bare partition for the window of one; metadata is carried onto every event and span of the run; a run_id is generated when absent.

The sync bridge¶

il.run(coro) drives any framework coroutine to completion from synchronous code. Unlike asyncio.run, it works where a loop is already running (Jupyter) and reuses one persistent background loop across calls, so loop-bound state such as an AsyncRESTClient cached on a connection stays valid from one call to the next. Ctrl-C cancels the coroutine. Calling it from code already on that loop raises RuntimeError; await there instead.

asset.run(), asset.materialize() and dag.materialize() are built on it.

Runners¶

Companion packages register Docker and Kubernetes runners under the same interface, each operation in its own container or Job; see Ecosystem.

A runner is configuration. Each runner.run(dag) makes a run from it that walks the DAG once, so one runner serves any number of runs, concurrently too. Operations are scheduled dynamically: as soon as every predecessor has completed, an operation becomes ready and starts while a slot is free. A failed operation cancels everything downstream of it.

Options¶

Runners are Serializable, so they can be configured from settings: il.Runner.from_settings(settings.runner) resolves runner.type in the RUNNERS registry and constructs it with runner.config.

Results¶

result.status              # ExecutionStatus.COMPLETED or FAILED
result.window
result.execution_time      # seconds
result.executions          # id -> ExecutionInfo
result.completed_ids, result.failed_ids, result.canceled_ids
print(result)
# RunResult(status=failed, window=2026-01-15, completed=2, failed=1, canceled=1, time=1.20s, failed=[...], canceled=[...])

ExecutionInfo carries component_id, component_key, status, attempt, start_time, end_time, execution_time, and result: the OperationResult the operation came to, with its effects (what it asked the platform to persist) and, on failure, its error and traceback, which ExecutionInfo.error and ExecutionInfo.traceback read. ExecutionStatus is QUEUED, READY, RUNNING, COMPLETED, FAILED, SKIPPED (disabled), CANCELED (downstream of a failure).

Failure handling¶

Asset failures are absorbed into the result: the node is marked failed with its message and traceback, its dependents are canceled, and the run continues or stops depending on fail_fast. With reraise=True the original exception is re-raised once the run is finalized, so events and results are complete before it surfaces. A failure of the walk machinery itself (a deadlock, an invalid graph) raises RunnerError.

Before anything executes, the runner asks every enabled operation's partitioning to check the window and lists every failure in one PartitionError, so a run no asset can serve fails without executing anything.

Retrying¶

An operation that fails is attempted again when it carries a RetryPolicy, without the run seeing the intermediate failure. Nothing retries by default: a policy is in force only where a component declares one.

@il.asset(retry=il.RetryPolicy(max_attempts=3, delay=5))
def orders(**kwargs): ...

The policy is declared on the component whose unit it governs, so one budget never means several things:

max_attempts counts the first attempt, and the wait before each further one grows geometrically from delay by backoff, capped at max_delay and spread by jitter so operations failing together do not retry in lockstep. Whether a given error is worth another attempt is behaviour rather than configuration: override retryable() to recognise one that never will be.

class Orders(il.Asset):
    retry = il.RetryPolicy(max_attempts=3)

    def retryable(self, error: Exception) -> bool:
        return not isinstance(error, PermissionError)

A retried attempt is not a verdict. The node keeps its status, nothing downstream is canceled, and fail_fast is not tripped; only an exhausted or declined failure marks the node failed. Each attempt is recorded as an operation_retried event, and operation_failed means the budget ran out.

A job's policy works one level up: when a run fails, the platform queues the next attempt after the backoff. Those attempts form a stack, and the rest of the system reads the stack rather than any one attempt. An automatic retry always re-runs only what failed, carrying forward every operation that already succeeded anywhere in the stack; the manual retry endpoint additionally offers re-running everything, for when the earlier success is the thing you distrust. Either way a retry continues from the stack's latest attempt: retrying an earlier attempt retries the stack, and a stack whose latest attempt is still pending, or succeeded, has nothing to retry.

Running single assets¶

asset.run() and asset.materialize() bypass the runner. Pass the DAG when the asset has dependencies, so upstream data can be read:

dag = il.DAG(source)
source.report.materialize(partition, dag=dag)