Execution¶

A DAG orders the operations of a run; a runner walks it and reports the result.

DAG ¶

DAG(*items: Workload | type[Workload])

Directed acyclic graph of operations.

Edges come from the nodes' own bindings: whatever fills an asset-kind relation (see upstream_relations) is a node this one runs after. The DAG validates every live node's relations and provides topological ordering for parallel execution.

Create a DAG from workloads (assets, sources, jobs, ...).

Parameters:

Name	Type	Description	Default
`*items`	`Workload \| type[Workload]`	Workload instances or classes to include; each flattens into the operations it provides.	required

from_paths ¶

from_paths(paths: list[str]) -> DAG

Build a DAG from dotted import paths, one Source or Asset class each.

Parameters:

Name	Type	Description	Default
`paths`	`list[str]`	Dotted (or composite `module:Class`) import paths.	required

Returns:

Type	Description
`DAG`	The DAG over one default instance of each class.

Raises:

Type	Description
`DAGError`	If a path fails to import, or names something other than a Source or Asset class.

from_spec ¶

from_spec(
    spec: DAGSpec,
    catalog: Catalog | None = None,
    *,
    resolve: Callable[[str], Component] | None = None,
) -> DAG

Reconstruct a DAG from a spec.

Parameters:

Name	Type	Description	Default
`spec`	`DAGSpec`	A DAGSpec produced by `to_spec`.	required
`catalog`	`Catalog \| None`	Catalog used to resolve `key` references, shared across the spec's items. Defaults to the settings-configured catalog, built lazily.	`None`
`resolve`	`Callable[[str], Component] \| None`	Called with the id of a reference no item of the spec carries; `None` makes such a reference an error.	`None`

Returns:

Type	Description
`DAG`	A new DAG with the same structure.

from_spec_file ¶

from_spec_file(
    path: str | Path,
    catalog: Catalog | None = None,
    *,
    resolve: Callable[[str], Component] | None = None,
) -> DAG

Compile a runnable component spec file into a DAG.

Loads every Spec document the file holds (with ${VAR} env interpolation), reconstructs them against one shared registry so a reference may cross documents, and compiles the DAG over all of them: the file-based counterpart of targeting a runnable component by id, and what to_spec writes back.

Parameters:

Name	Type	Description	Default
`path`	`str \| Path`	Path to the YAML spec document(s).	required
`catalog`	`Catalog \| None`	Catalog used to resolve `key` references. Defaults to the settings-configured catalog, built lazily.	`None`
`resolve`	`Callable[[str], Component] \| None`	Called with the id of a reference no document carries; `None` makes such a reference an error.	`None`

Invalid documents surface as SpecError from the spec loader.

Returns:

Type	Description
`DAG`	The DAG over every root the file declares.

Raises:

Type	Description
`DAGError`	If a root's kind declares no workload.

get_predecessors ¶

get_predecessors(operation_id: str) -> list[str]

Return upstream dependency ids for the given operation.

Parameters:

Name	Type	Description	Default
`operation_id`	`str`	Id of the operation to look up.	required

Raises:

Type	Description
`AssetNotFoundError`	If the operation id is not in the DAG.

get_successors ¶

get_successors(operation_id: str) -> list[str]

Return downstream dependent ids for the given operation.

Parameters:

Name	Type	Description	Default
`operation_id`	`str`	Id of the operation to look up.	required

Raises:

Type	Description
`AssetNotFoundError`	If the operation id is not in the DAG.

materialize ¶

materialize(
    window: Partition | PartitionWindow | None = None,
) -> RunResult

Execute all operations in dependency order using a default AsyncRunner.

Sync entrypoint for scripts, REPLs, and notebooks; drives materialize_async to completion on the bridge loop (see interloper.run):

result = dag.materialize(partition)

Async code awaits materialize_async instead.

Parameters:

Name	Type	Description	Default
`window`	`Partition \| PartitionWindow \| None`	The window every operation in the DAG is run for; a partition is accepted as the window of one. `None` for an unpartitioned DAG.	`None`

Returns:

Type	Description
`RunResult`	The result of the DAG execution.

materialize_async ¶

materialize_async(
    window: Partition | PartitionWindow | None = None,
) -> RunResult

Execute all operations in dependency order using a default AsyncRunner.

Parameters:

Name	Type	Description	Default
`window`	`Partition \| PartitionWindow \| None`	The window every operation in the DAG is run for; a partition is accepted as the window of one. `None` for an unpartitioned DAG.	`None`

Returns:

Type	Description
`RunResult`	The result of the DAG execution.

mini_dag ¶

mini_dag(operation_id: str) -> DAG

Create a mini-DAG with the target operation and its immediate parents.

A DAG over the target alone: its bound upstreams join as disabled copies through the very mechanism any run uses for an upstream it does not materialize (_include_read_only_upstreams), so the parents are there, under their own ids, read instead of executed.

Parameters:

Name	Type	Description	Default
`operation_id`	`str`	Id of the operation the mini-DAG is built around.	required

Returns:

Type	Description
`DAG`	A new DAG containing only the target operation and its parents.

Raises:

Type	Description
`AssetNotFoundError`	If the operation id is not in the DAG.

to_spec ¶

to_spec() -> DAGSpec

Serialize this DAG to a reconstructible spec.

One item per root of the graph: a standalone asset is an item of its own, and a source-owned asset travels inside its owning source's item, which is what also gives the parent of an upstream this run only reads an item of its own, carrying that one asset and nothing else of the source.

Every item shares one closed SerializationContext over the DAG's own nodes: a destination bound to several roots is written once and referenced everywhere else, a source writes the graph's copies of its assets (flagged read-only in a mini-DAG, or only some of them in a partial run) rather than its originals, and a binding pointing at an owned component outside this graph is left out rather than written as a reference nothing answers.

Returns:

Type	Description
`DAGSpec`	A DAGSpec that can reconstruct an equivalent DAG.

topological_generations ¶

topological_generations() -> list[list[Operation]]

Return operations grouped by parallelizable generations.

Each inner list contains operations that can be executed in parallel. Lists are ordered so that all dependencies of a level appear in previous levels (Kahn's algorithm).

Only enabled operations appear in the generations. Edges from disabled operations count as already satisfied, mirroring the runners, which mark those nodes as skipped (e.g. the parents in a mini_dag).

Returns:

Type	Description
`list[list[Operation]]`	A list of operation groups ordered by dependency level.

Raises:

Type	Description
`CircularDependencyError`	If a cycle is detected.

Runner ¶

Runner(**data: Any)

Bases: interloper.serializable.base.Serializable

Configuration a run is made from: how many operations run at once, where, and what a failure does.

A runner holds nothing per run, so one runner serves any number of runs, concurrently too. Each kind of runner names its kind of Run, which owns the walk, what the run holds, and where an operation executes. await run from async code, or drive it from a sync entrypoint with interloper.run:

result = await runner.run(dag)     # async
result = il.run(runner.run(dag))   # sync edge

capacity ¶

capacity

How many operations a run executes at once.

key ¶

key = 'runner'

str(object='') -> str str(bytes_or_buffer[, encoding[, errors]]) -> str

Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.str() (if defined) or repr(object). encoding defaults to sys.getdefaultencoding(). errors defaults to 'strict'.

model_config ¶

model_config = {'arbitrary_types_allowed': True, 'ignored_types': (<class 'interloper.serializable.base.IgnoredDescriptor'>,)}

dict() -> new empty dictionary dict(mapping) -> new dictionary initialized from a mapping object's (key, value) pairs dict(iterable) -> new dictionary initialized as if via: d = {} for k, v in iterable: d[k] = v dict(**kwargs) -> new dictionary initialized with the name=value pairs in the keyword argument list. For example: dict(one=1, two=2)

from_settings ¶

from_settings(settings: RunnerSettings) -> Self

Construct the runner the settings describe.

Resolves settings.type in RUNNERS and constructs it with settings.config. Called on a subclass, the resolved runner must be of that subclass.

Parameters:

Name	Type	Description	Default
`settings`	`RunnerSettings`	Runner settings carrying the registry key (`type`) and the keyword arguments to construct it with (`config`).	required

Returns:

Type	Description
`Self`	The configured runner instance.

Raises:

Type	Description
`ConfigError`	If no runner is registered under `settings.type`, or the registered class is not of the receiving class.

run ¶

run(
    dag: DAG,
    window: Partition | PartitionWindow | None = None,
    metadata: dict[str, Any] | None = None,
) -> RunResult

Run the DAG and return the result.

Events carrying the run's id reach on_event (when set) before this returns: the EventBus is process-wide, so several runs in one process would otherwise reach every handler. The run is traced as interloper.runner.run, under the parent span the metadata or the environment carries. Operation failures are recorded on the result; with reraise set, the first one is raised once the run has finished.

Parameters:

Name	Type	Description	Default
`dag`	`DAG`	The DAG to run.	required
`window`	`Partition \| PartitionWindow \| None`	The window the run covers; a partition is accepted as the window of one.	`None`
`metadata`	`dict[str, Any] \| None`	Run-level metadata (`run_id`, `backfill_id`, ...); a `run_id` is generated when absent.	`None`

Returns:

Type	Description
`RunResult`	The run's result.

AsyncRunner ¶

AsyncRunner(**data: Any)

Bases: interloper.runner.base.Runner

Run every operation in this process: max_workers at a time, sync ones offloaded to threads.

AsyncRunner(max_workers=1) is serial and deterministic (see SerialRunner); the default runs four operations at once:

result = await AsyncRunner(max_workers=2, on_event=log_event).run(dag)   # async
result = il.run(AsyncRunner(on_event=log_event).run(dag))                 # sync edge

capacity ¶

capacity

How many operations a run executes at once: max_workers.

key ¶

key = 'async_runner'

str(object='') -> str str(bytes_or_buffer[, encoding[, errors]]) -> str

Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.str() (if defined) or repr(object). encoding defaults to sys.getdefaultencoding(). errors defaults to 'strict'.

model_config ¶

model_config = {'arbitrary_types_allowed': True, 'ignored_types': (<class 'interloper.serializable.base.IgnoredDescriptor'>,)}

dict() -> new empty dictionary dict(mapping) -> new dictionary initialized from a mapping object's (key, value) pairs dict(iterable) -> new dictionary initialized as if via: d = {} for k, v in iterable: d[k] = v dict(**kwargs) -> new dictionary initialized with the name=value pairs in the keyword argument list. For example: dict(one=1, two=2)

Run ¶

Run(
    runner: Runner,
    dag: DAG,
    window: PartitionWindow | None,
    metadata: dict[str, Any],
)

Bases: interloper.runner.run.Run

One run of an AsyncRunner: every operation executes in this process, on its event loop.

A sync data() is offloaded to a thread by the asset itself, so one event loop serves the whole run.

execute ¶

execute(operation: Operation) -> OperationResult

Execute the operation here, under its retry policy, reporting each retry as it happens.

Parameters:

Name	Type	Description	Default
`operation`	`Operation`	The operation to execute.	required

Returns:

Type	Description
`OperationResult`	Its result.

SerialRunner ¶

SerialRunner(**data: Any)

Bases: interloper.runner.async_runner.AsyncRunner

Execute operations one at a time in dependency order.

The simplest runner: deterministic, easy to debug. It is the async engine with a single concurrency slot, so there is never more than one operation in flight:

result = await SerialRunner(on_event=log_event).run(dag)
# or, from a sync edge: il.run(SerialRunner().run(dag))

key ¶

key = 'serial_runner'

str(object='') -> str str(bytes_or_buffer[, encoding[, errors]]) -> str

Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.str() (if defined) or repr(object). encoding defaults to sys.getdefaultencoding(). errors defaults to 'strict'.

model_config ¶

model_config = {'arbitrary_types_allowed': True, 'ignored_types': (<class 'interloper.serializable.base.IgnoredDescriptor'>,)}

dict() -> new empty dictionary dict(mapping) -> new dictionary initialized from a mapping object's (key, value) pairs dict(iterable) -> new dictionary initialized as if via: d = {} for k, v in iterable: d[k] = v dict(**kwargs) -> new dictionary initialized with the name=value pairs in the keyword argument list. For example: dict(one=1, two=2)

RunResult ¶

RunResult(window: PartitionWindow | None = None, status: ExecutionStatus = <ExecutionStatus.COMPLETED: 'completed'>, executions: dict[str, ExecutionInfo] = <factory>, execution_time: float = 0.0)

Result of a single DAG execution (one partition, window, or unpartitioned).

canceled_ids ¶

canceled_ids

Ids of the operations that were canceled (downstream of a failure).

completed_ids ¶

completed_ids

Ids of the operations that completed successfully.

execution_time ¶

execution_time = 0.0

Convert a string or number to a floating point number, if possible.

failed_ids ¶

failed_ids

Ids of the operations that failed.

status ¶

status = <ExecutionStatus.COMPLETED: 'completed'>

Execution status for operations and runs.

Workload ¶

Bases: abc.ABC

What a run may target: a provider of operations.

billable declares whether runs of this workload count against the org's run quota; the platform records it on each run at creation so billing stays correct even after the component is deleted.

billable ¶

billable = True

bool(x) -> bool

Returns True when the argument x is true, False otherwise. The builtins True and False are the only two instances of the class bool. The class bool is a subclass of the class int, and cannot be subclassed.

operations ¶

operations() -> list[Operation]

The operations this workload flattens into.

Returns:

Type	Description
`list[Operation]`	The operations, before any graph assembly.

Operation ¶

Operation(**data: Any)

Bases: interloper.component.base.Component, interloper.operation.base.Workload

A unit of work: the node a DAG orders and a runner drives.

An operation is a component: the DAG, the runner and the platform all address it by its component identity, and its events and executions are keyed by its row id. What it adds to a component is the execution contract (execute, failure) and the node attributes the graph machinery reads, which Asset (the graph-structured, partitioned operation) narrows with its own fields and properties.

kind is deliberately empty: an operation is a contract several kinds satisfy, not a kind of its own, and declaring it here is what stops Component.__init_subclass__ deriving one. Each implementor declares its own.

enabled is the same word Job and Hook already use for "this will run": a disabled operation stays in the graph, ordering whatever depends on it, and does not execute. Asset adds the half only an asset has, which is that a disabled one is read-only and its downstreams resolve its data from its destination instead of recomputing it. enabled and retry are inherited: an owned operation reads them through its owner, which decides (a source's assets run only while the source is enabled; one left without a retry policy uses its owner's).

capture_traceback controls whether a failed execution's traceback is attached to its failure event; off for operations whose raw errors embed secrets (credential exchanges carry them in URLs).

reads_upstream_data says whether executing the node reads its upstreams' data for the run's partition. An asset does; a view or a reference does not, which is what lets them sit, unpartitioned, on partitioned upstreams: the partition rule in the DAG only applies to nodes that read.

capture_traceback ¶

capture_traceback = True

bool(x) -> bool

Returns True when the argument x is true, False otherwise. The builtins True and False are the only two instances of the class bool. The class bool is a subclass of the class int, and cannot be subclassed.

enabled ¶

enabled = <interloper.component.inherited.InheritedSetting object at 0x7febf2404fe0>

The read-through descriptor installed over one inherited field.

A data descriptor, so it wins over the instance __dict__ on reads while Pydantic keeps writing there: the stored value is the override, and the read is the effective value.

key ¶

key = 'operation'

str(object='') -> str str(bytes_or_buffer[, encoding[, errors]]) -> str

Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.str() (if defined) or repr(object). encoding defaults to sys.getdefaultencoding(). errors defaults to 'strict'.

kind ¶

kind = ''

str(object='') -> str str(bytes_or_buffer[, encoding[, errors]]) -> str

Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.str() (if defined) or repr(object). encoding defaults to sys.getdefaultencoding(). errors defaults to 'strict'.

model_config ¶

model_config = {'arbitrary_types_allowed': True, 'ignored_types': (<class 'interloper.serializable.base.IgnoredDescriptor'>, <class 'interloper.component.relation.Relation'>)}

dict() -> new empty dictionary dict(mapping) -> new dictionary initialized from a mapping object's (key, value) pairs dict(iterable) -> new dictionary initialized as if via: d = {} for k, v in iterable: d[k] = v dict(**kwargs) -> new dictionary initialized with the name=value pairs in the keyword argument list. For example: dict(one=1, two=2)

reads_upstream_data ¶

reads_upstream_data = True

bool(x) -> bool

Returns True when the argument x is true, False otherwise. The builtins True and False are the only two instances of the class bool. The class bool is a subclass of the class int, and cannot be subclassed.

relations ¶

relations = {}

dict() -> new empty dictionary dict(mapping) -> new dictionary initialized from a mapping object's (key, value) pairs dict(iterable) -> new dictionary initialized as if via: d = {} for k, v in iterable: d[k] = v dict(**kwargs) -> new dictionary initialized with the name=value pairs in the keyword argument list. For example: dict(one=1, two=2)

retry ¶

retry = <interloper.component.inherited.InheritedSetting object at 0x7febf2480800>

The read-through descriptor installed over one inherited field.

A data descriptor, so it wins over the instance __dict__ on reads while Pydantic keeps writing there: the stored value is the override, and the read is the effective value.

effective_window ¶

effective_window(
    window: PartitionWindow | None,
) -> PartitionWindow | None

The window this operation actually consumes.

Unpartitioned operations ignore any window the run was given.

Parameters:

Name	Type	Description	Default
`window`	`PartitionWindow \| None`	The window the run was scoped to, `None` when it was unscoped.	required

Returns:

Type	Description
`PartitionWindow \| None`	The window unchanged for a partitioned operation, `None` otherwise.

execute ¶

execute(context: OperationContext) -> OperationResult

Execute this operation.

Parameters:

Name	Type	Description	Default
`context`	`OperationContext`	The facts this execution is scoped to.	required

Returns:

Type	Description
`OperationResult`	The effects to persist (often none).

failure ¶

failure(error: Exception) -> OperationResult

Describe a failed execution in terms the platform can persist.

Consulted by the runner when execute raises. Override to curate the message or attach state effects (e.g. a retry slot); the default formats the error with pydantic input values stripped.

Parameters:

Name	Type	Description	Default
`error`	`Exception`	The exception `execute` raised.	required

Returns:

Type	Description
`OperationResult`	A result carrying the message and any state effects.

model_post_init ¶

model_post_init(context: Any) -> None

Default id to a generated UUID if not provided.

Parameters:

Name	Type	Description	Default
`context`	`Any`	The pydantic validation context, unused here.	required

operations ¶

operations() -> list[Operation]

An operation is trivially its own workload.

Returns:

Type	Description
`list[Operation]`	A single-element list holding this operation.

retryable ¶

retryable(error: Exception) -> bool

Whether another attempt at this operation is worth making.

Consulted by the runner before it spends an attempt from retry. Override to recognise a permanent error, such as a vendor rejecting a request it will reject identically every time. The default is permissive: the budget, not the classifier, is what bounds waste.

Parameters:

Name	Type	Description	Default
`error`	`Exception`	The exception `execute` raised.	required

Returns:

Type	Description
`bool`	`True` when the failure may be transient.

upstream_relations ¶

upstream_relations() -> dict[str, Relation]

The node's relations that another node in the graph fills.

An asset-kind relation is an edge: whatever fills it is a node the graph must order before this one. Every other relation points at something outside the graph (a connection, a config, a destination).

Returns:

Type	Description
`dict[str, Relation]`	Relation name to declaration, for the relations accepting assets.

OperationContext ¶

OperationContext(window: PartitionWindow | None = None, dag: DAG | None = None, metadata: dict[str, Any] = <factory>)

The facts an operation executes with.

Handed down by the runner, one per node execution: the operation-level counterpart of the asset-level ExecutionContext. dag is the graph the node belongs to, how an operation reaches its upstream nodes' outputs; operations without dependencies ignore it.

OperationResult ¶

OperationResult(error: str | None = None, config: dict[str, Any] = <factory>, state: dict[str, Any] = <factory>, traceback: str | None = None, exception: BaseException | None = None)

What an operation came to: the effects the platform persists, and the verdict.

config fields are merged into the target component's stored config (through its encryption path) and state fields onto its machine-owned state: how an operation's effects reach the row without the operation knowing the store. An operation's execute returns one carrying its effects, and its failure hook one carrying the persistence-safe error. The attempt loop (from_execution) completes a failure with the traceback (when the operation's class allows one) and the original exception, kept in memory only so a runner can re-raise it. An unset error means the operation succeeded.

succeeded ¶

succeeded

Whether the operation succeeded, which is whether no error is set.

Returns:

Type	Description
	`True` when `error` is unset.

from_execution ¶

from_execution(
    operation: Operation,
    context: OperationContext,
    *,
    on_retry: Callable[[str], None],
    on_restart: Callable[[], None],
) -> OperationResult

Execute an operation under its retry policy and return what it came to.

Each attempt runs inside the interloper.operation.execute span. A failure is retried when the operation's RetryPolicy allows another attempt and Operation.retryable agrees: on_retry is told the error as it happens, the policy's backoff is slept, on_restart is told the next attempt starts, and the operation executes again. Any other failure becomes the operation's failure result, completed with the formatted error when the hook left it empty, the traceback when the class's capture_traceback allows, and the original exception.

Parameters:

Name	Type	Description	Default
`operation`	`Operation`	The operation to execute.	required
`context`	`OperationContext`	The facts it executes with.	required
`on_retry`	`Callable[[str], None]`	Told the formatted error of each attempt that will be retried, before its backoff.	required
`on_restart`	`Callable[[], None]`	Told when the next attempt starts, after the backoff.	required

Returns:

Type	Description
`OperationResult`	The result: the operation's own on success, its failure otherwise.