loom.etl.runner

Runner API for compiling and executing ETL pipelines.

class loom.etl.runner.ETLRunner(reader, writer, observability=None, dispatcher=None, checkpoint_store=None, client_executor=None, config_context=None, agents=None)[source]

Bases: object

Wire storage config, backend I/O, and executor into one callable.

Parameters:

  • reader (SourceReader) – Source reader implementation.

  • writer (TargetWriter) – Target writer implementation.

  • observers – Lifecycle observers wrapped in a composite observer.

  • dispatcher (ParallelDispatcher | None) – Parallel task dispatcher.

  • config_context (ConfigContext | None) – Config the steps’ FromConfig values come from. run() checks every declared key against it before any step runs, and each step resolves its values when it executes.

  • agents (AgentBatchRunner | None) – Runner of the agents the steps’ WithAgent declare. run() checks every declaration against it before any step runs.

  • observability (ObservabilityRuntime | None)

  • checkpoint_store (CheckpointStore | None)

  • client_executor (ClientCommandExecutor | None)

classmethod from_config(config, obs_config=None, *, spark=None, dispatcher=None, cleaner=None, extra_observers=None, config_context=None, agents=None)[source]

Build an ETLRunner from resolved config objects.

Parameters:

  • extra_observers (Sequence[LifecycleObserver] | None) – Optional additional LifecycleObserver instances appended after the config-derived ones. Use this to inject orchestrator-specific observers (e.g. the Prefect TaskRun observer) without subclassing the runner.

  • config_context (ConfigContext | None) – Config the steps’ FromConfig values come from.

  • agents (AgentBatchRunner | None) – Runner of the agents the steps’ WithAgent declare. On Spark every declaration is refused at compile time with AGENT_UNSUPPORTED_ENGINE: agents map Polars frames only.

  • config (StorageConfig)

  • obs_config (ETLObservabilityConfig | None)

  • spark (SparkSession | None)

  • dispatcher (ParallelDispatcher | None)

  • cleaner (TempCleaner | None)

Return type:

ETLRunner

classmethod from_yaml(path, *, resolvers=(), spark=None, dispatcher=None, extra_observers=None)[source]

Load config from a YAML file and build an ETLRunner.

storage.tables and storage.files accept the list form or a mapping keyed by logical name, and storage.tables, storage.files and storage.profiles are merged by key across includes, with duplicate keys reported as an error.

The whole file backs the steps’ FromConfig values, so a step may read a key outside storage: (respondio.api_token); it is resolved, with its interpolations and resolvers, when the step runs.

An ai: section serves the steps’ WithAgent declarations; its specs globs resolve against ai.root, a path relative to the directory holding the YAML (the working directory for a cloud URI), or against that directory when ai.root is absent, and may not leave it.

Parameters:

  • resolvers (Sequence[ConfigResolver]) – Resolvers for ${name:key} placeholders, registered before the built-in secrets and ssm defaults. A resolver named like a default replaces it.

  • extra_observers (Sequence[LifecycleObserver] | None) – Optional additional LifecycleObserver instances forwarded to from_config().

  • path (str)

  • spark (SparkSession | None)

  • dispatcher (ParallelDispatcher | None)

Return type:

ETLRunner

classmethod from_spark(spark, obs_config=None, *, dispatcher=None, cleaner=None)[source]

Build an ETLRunner using Spark as the execution engine.

Parameters:

  • spark (SparkSession)

  • obs_config (ETLObservabilityConfig | None)

  • dispatcher (ParallelDispatcher | None)

  • cleaner (TempCleaner | None)

Return type:

ETLRunner

classmethod from_dict(storage, observability=None, *, spark=None, dispatcher=None, cleaner=None)[source]

Build an ETLRunner from pre-resolved plain Python dicts.

storage.tables and storage.files accept the list form or a mapping keyed by logical name. A path.profile: <name> key is replaced by the fields of storage.profiles[<name>] the path does not set itself.

Raises:

  • ValueError – When a mapping entry carries a name different from its key, a path.profile is not declared in storage.profiles, or a profile carries uri.

  • msgspec.ValidationError – When the storage shape is invalid, including a profile with an unknown field.

Parameters:

  • storage (dict[str, Any])

  • observability (dict[str, Any] | None)

  • spark (SparkSession | None)

  • dispatcher (ParallelDispatcher | None)

  • cleaner (TempCleaner | None)

Return type:

ETLRunner

run(pipeline, params, *, include=None, correlation_id=None, run_id=None, attempt=1, last_attempt=True)[source]

Compile, optionally filter, and execute pipeline.

With a config context, compilation checks every FromConfig key of the pipeline before any step runs.

Parameters:

  • pipeline (type[ETLPipeline[Any]]) – ETLPipeline class to compile and execute.

  • params (Any) – ETLParams instance for this run.

  • include (Sequence[str] | None) – If set, only steps whose names are in this sequence run.

  • correlation_id (str | None) – Logical business unit identifier, stable across retries.

  • run_id (str | None) – Execution-specific identifier for lineage records. If None, a random UUID is generated. Callers that share a traceability identifier with an orchestrator (e.g. Prefect) should pass it explicitly so lineage and orchestrator records align.

  • attempt (int) – Current attempt number (1-based).

  • last_attempt (bool) – Whether this is the final allowed attempt.

Return type:

None

flush()[source]

Flush buffered ETL observability sinks after a run.

Return type:

None

cleanup_correlation(correlation_id)[source]

Remove all CORRELATION-scope intermediates for correlation_id.

Parameters:

correlation_id (str)

Return type:

None

exception loom.etl.runner.InvalidStageError(unknown=None, *, include=None)[source]

Bases: ValueError

Raised when a selection names a step or process absent from the compiled plan.

Parameters:

  • unknown (frozenset[str] | None) – The selected names that match no step or process.

  • include (frozenset[str] | None) – Deprecated keyword spelling of unknown, removed in loom 3.0.

Return type:

None

unknown

The selected names that match no step or process.

Raises:

TypeError – When both or neither of unknown and include are given.

Parameters:
Return type:

None