loom.etl.compiler¶

ETL compiler public API.

class loom.etl.compiler.ETLCompiler(catalog=None, config_context=None, agents=None)[source]¶

Bases: object

Compiles ETL declarations into immutable execution plans.

Compilation is idempotent per class — results are cached. Call compile() with the top-level ETLPipeline subclass to obtain a PipelinePlan.

Parameters:

catalog (TableDiscovery | None) – Optional TableDiscovery instance. When provided, all TABLE sources and IntoTable targets are validated against the catalog at compile time.
config_context (ConfigContext | None) – Optional config the runner was built from. When provided, every FromConfig key is resolved and validated at compile time; values are discarded, never stored on a plan.
agents (AgentBatchRunner | None) – Optional agent runner. When provided, or when a config context is, every WithAgent is validated through it at compile time; without one, a declared agent cannot be served.

Example:

plan = ETLCompiler().compile(DailyOrdersPipeline)
plan = ETLCompiler(catalog=HiveCatalog()).compile(DailyOrdersPipeline)

compile(pipeline_type)[source]¶

Compile a pipeline and return its immutable plan.

Parameters:: pipeline_type (type[ETLPipeline[Any]]) – Concrete ETLPipeline subclass.
Returns:: Fully validated PipelinePlan.
Raises:: ETLCompilationError – If any structural constraint is violated, a FromConfig key does not resolve against the config context, or a WithAgent cannot be served.
Return type:: PipelinePlan

compile_process(process_type)[source]¶

Compile a single process.

Parameters:: process_type (type[ETLProcess[Any]]) – Concrete ETLProcess subclass.
Returns:: Validated ProcessPlan.
Raises:: ETLCompilationError – If any structural constraint is violated, a FromConfig key does not resolve against the config context, or a WithAgent cannot be served.
Return type:: ProcessPlan

compile_step(step_type)[source]¶

Compile a single step.

Parameters:: step_type (type[ETLStep[Any]]) – Concrete ETLStep subclass.
Returns:: Validated StepPlan.
Raises:: ETLCompilationError – If any structural constraint is violated, a FromConfig key does not resolve against the config context, or a WithAgent cannot be served.
Return type:: StepPlan

exception loom.etl.compiler.ETLCompilationError(code, component, message, field=None)[source]¶

Bases: Exception

Raised when an ETL class fails structural validation.

Parameters:

code (ETLErrorCode)
component (str)
message (str)
field (str | None)

Return type:

None

code¶: Machine-readable ETLErrorCode.

component¶: Qualified name of the step or component that failed.

field¶: Optional name of the specific field or parameter involved.

Use the factory class-methods to construct instances — they own all message formatting so call-sites stay readable.

Example:

raise ETLCompilationError.upsert_no_keys(MyStep)

classmethod invalid_params_type(step, expected)[source]¶

First execute() parameter has wrong type annotation.

Parameters:

step (type)
expected (type)

Return type:

ETLCompilationError

classmethod invalid_params_name(step, got)[source]¶

First execute() parameter is not named ‘params’.

Parameters:

step (type)
got (str)

Return type:

ETLCompilationError

classmethod missing_source_params(step, missing)[source]¶

Sources declared in sources but absent as kw-only execute() parameters.

Parameters:

step (type)
missing (frozenset[str])

Return type:

ETLCompilationError

classmethod extra_source_params(step, extra)[source]¶

Kw-only execute() parameters not declared in sources nor as FromConfig.

Parameters:

step (type)
extra (frozenset[str])

Return type:

ETLCompilationError

classmethod missing_params_fields(component, missing, context)[source]¶

Component requires params fields that the context params type lacks.

Parameters:

component (type)
missing (frozenset[str])
context (type)

Return type:

ETLCompilationError

classmethod unknown_source_table(step, alias, table_ref)[source]¶

Source binding references a table not in the catalog.

Parameters:

step (type)
alias (str)
table_ref (str)

Return type:

ETLCompilationError

classmethod unknown_target_table(step, table_ref)[source]¶

Target references a table not in the catalog.

Parameters:

step (type)
table_ref (str)

Return type:

ETLCompilationError

classmethod temp_not_produced(step, alias, temp_name)[source]¶

FromTemp references a temp name with no prior IntoTemp.

Parameters:

step (type)
alias (str)
temp_name (str)

Return type:

ETLCompilationError

classmethod upsert_no_keys(step, method='upsert')[source]¶

upsert()/update() declared with no key columns.

Parameters:

step (type)
method (str)

Return type:

ETLCompilationError

classmethod upsert_exclude_include_conflict(step, method='upsert')[source]¶

upsert()/update() has both exclude= and include= set.

Parameters:

step (type)
method (str)

Return type:

ETLCompilationError

classmethod upsert_key_in_exclude(step, overlap, method='upsert')[source]¶

upsert()/update() exclude= contains columns that are also merge keys.

Parameters:

step (type)
overlap (frozenset[str])
method (str)

Return type:

ETLCompilationError

classmethod update_empty_update_set(step, include)[source]¶

update() include= is fully absorbed by keys and partition columns.

Parameters:

step (type)
include (tuple[str, ...])

Return type:

ETLCompilationError

classmethod missing_generic_param(cls_type, kind)[source]¶

ETLStep/ETLProcess/ETLPipeline used without a generic parameter.

Parameters:

cls_type (type)
kind (str)

Return type:

ETLCompilationError

classmethod missing_target(step)[source]¶

‘target’ is required but not declared on the step.

Parameters:: step (type)
Return type:: ETLCompilationError

classmethod invalid_target_type(step)[source]¶

‘target’ is not an IntoTable/IntoFile/IntoTemp instance.

Parameters:: step (type)
Return type:: ETLCompilationError

classmethod invalid_sources_type(step)[source]¶

‘sources’ is not a Sources(…) or SourceSet instance.

Parameters:: step (type)
Return type:: ETLCompilationError

classmethod invalid_process_item(pipeline, item)[source]¶

An item in pipeline.processes is not an ETLProcess subclass.

Parameters:

pipeline (type)
item (object)

Return type:

ETLCompilationError

classmethod invalid_step_item(process, item)[source]¶

An item in process.steps is not an ETLStep subclass.

Parameters:

process (type)
item (object)

Return type:

ETLCompilationError

classmethod duplicate_temp_name(step, name)[source]¶

Two steps write to the same IntoTemp name without append=True.

Parameters:

step (type)
name (str)

Return type:

ETLCompilationError

classmethod invalid_temp_append_mix(step, name)[source]¶

A step mixes append=True and append=False on the same IntoTemp name.

Parameters:

step (type)
name (str)

Return type:

ETLCompilationError

classmethod unknown_template_field(step, field_name, path, params_type)[source]¶

A file path template references a field not declared on the params type.

Parameters:

step (type)
field_name (str)
path (str)
params_type (type)

Return type:

ETLCompilationError

classmethod unknown_param_field(step, field_name, params_type)[source]¶

A ParamExpr references a field not declared on the params type.

Parameters:

step (type)
field_name (str)
params_type (type)

Return type:

ETLCompilationError

classmethod missing_config_params(step, missing, marker='FromConfig')[source]¶

FromConfig or WithAgent attributes absent as kw-only execute() parameters.

Parameters:

step (type)
missing (frozenset[str])
marker (str)

Return type:

ETLCompilationError

classmethod config_alias_conflict(step, aliases, marker='FromConfig')[source]¶

A FromConfig or WithAgent attribute shares its name with a source or the client.

Parameters:

step (type)
aliases (frozenset[str])
marker (str)

Return type:

ETLCompilationError

classmethod unsupported_config_value(step, marker='FromConfig')[source]¶

A StepSQL declares FromConfig or WithAgent values it has no execute() to receive.

Parameters:

step (type)
marker (str)

Return type:

ETLCompilationError

classmethod unresolved_config_value(step, alias, reason)[source]¶

A FromConfig key cannot be supplied from the runner’s config.

reason must never carry the value: callers pass the message of a ConfigValueError.

Parameters:

step (type)
alias (str)
reason (str)

Return type:

ETLCompilationError

classmethod agent_not_found(step, alias, reason)[source]¶

A WithAgent names an agent no configured artifact declares.

Parameters:

step (type)
alias (str)
reason (str)

Return type:

ETLCompilationError

classmethod agent_runner_not_configured(step, alias)[source]¶

A WithAgent is declared on a runner configured without an agent runner.

Parameters:

step (type)
alias (str)

Return type:

ETLCompilationError

classmethod agent_compilation_failed(step, alias, reason)[source]¶

The artifact a WithAgent names does not compile; reason carries its issues.

Parameters:

step (type)
alias (str)
reason (str)

Return type:

ETLCompilationError

classmethod agent_output_mismatch(step, alias, reason)[source]¶

A WithAgent output differs from the type_ref output of its artifact.

Parameters:

step (type)
alias (str)
reason (str)

Return type:

ETLCompilationError

classmethod agent_unpriced_budget(step, alias, reason)[source]¶

A budget is declared for an agent whose model has no known price.

Parameters:

step (type)
alias (str)
reason (str)

Return type:

ETLCompilationError

classmethod agent_budget_unenforceable(step, alias, reason)[source]¶

A step budget no worst case of one run can be reserved against.

Parameters:

step (type)
alias (str)
reason (str)

Return type:

ETLCompilationError

classmethod agent_unsupported_engine(step, alias, reason)[source]¶

A WithAgent declared on a runner whose engine no agent runner serves.

Parameters:

step (type)
alias (str)
reason (str)

Return type:

ETLCompilationError

classmethod agent_column_collision(step, alias, reason)[source]¶

An output field of a WithAgent is named like a run metadata column.

Parameters:

step (type)
alias (str)
reason (str)

Return type:

ETLCompilationError

class loom.etl.compiler.ETLErrorCode(value)[source]¶

Bases: StrEnum

Enumeration of all compiler validation failure codes.

class loom.etl.compiler.StepPlan(step_type, params_type, source_bindings, target_binding, streaming=False, config_bindings=(), agent_bindings=())[source]¶

Bases: object

Compiled plan for a single ETLStep.

Parameters:

step_type (type[Any]) – The concrete ETLStep subclass.
params_type (type[Any]) – The ParamsT generic argument.
source_bindings (tuple[SourceBinding, ...]) – Ordered source alias → spec bindings.
target_binding (TargetBinding) – Compiled target.
streaming (bool) – Whether the step opts into Polars streaming execution. Defaults to False.
config_bindings (tuple[ConfigValueBinding, ...]) – Ordered FromConfig alias → key bindings.
agent_bindings (tuple[AgentBinding, ...]) – Ordered WithAgent alias → agent bindings.

class loom.etl.compiler.ProcessPlan(process_type, params_type, nodes)[source]¶

Bases: object

Compiled plan for an ETLProcess.

Parameters:

process_type (type[Any]) – The concrete ETLProcess subclass.
params_type (type[Any]) – The ParamsT generic argument.
nodes (tuple[StepPlan | ParallelStepGroup, ...]) – Ordered sequence of step nodes (sequential or parallel groups).

class loom.etl.compiler.PipelinePlan(pipeline_type, params_type, nodes)[source]¶

Bases: object

Compiled plan for an ETLPipeline.

Parameters:

pipeline_type (type[Any]) – The concrete ETLPipeline subclass.
params_type (type[Any]) – The ParamsT generic argument.
nodes (tuple[ProcessPlan | ParallelProcessGroup, ...]) – Ordered sequence of process nodes.

class loom.etl.compiler.SourceBinding(alias, spec)[source]¶

Bases: object

Compiled binding between a source alias and its normalized spec.

Parameters:

alias (str) – Name matching the execute() keyword parameter.
spec (TableSourceSpec | FileSourceSpec | TempSourceSpec | MongoSourceSpec | ClickHouseSourceSpec | DynamoDbSourceSpec) – Normalized source specification.

class loom.etl.compiler.ConfigValueBinding(alias, key, value_type)[source]¶

Bases: object

Compiled binding between an execute() keyword and a config key.

Carries the declaration only; the value is resolved by the executor when the step runs and is never stored on a plan.

Parameters:

alias (str) – Name matching the execute() keyword parameter.
key (str) – Dot-separated config path.
value_type (object) – Type the resolved value is converted to.

class loom.etl.compiler.AgentBinding(alias, name, output_type, max_usd=None)[source]¶

Bases: object

Compiled binding between an execute() keyword and a declared agent.

Parameters:

alias (str) – Name matching the execute() keyword parameter.
name (str) – Agent name.
output_type (type[Any]) – Type every answer decodes into.
max_usd (Decimal | None) – Budget of one execution of the step, or None.

class loom.etl.compiler.TargetBinding(spec)[source]¶

Bases: object

Compiled binding for the step target.

Parameters:: spec (AppendSpec | ReplaceSpec | ReplacePartitionsSpec | ReplaceWhereSpec | UpsertSpec | UpdateSpec | HistorifySpec | FileSpec | TempSpec | TempFanInSpec | ClientSpec) – Normalized target specification.

class loom.etl.compiler.ParallelStepGroup(plans)[source]¶

Bases: object

A group of StepPlan items that execute concurrently.

Parameters:: plans (tuple[StepPlan, ...]) – Step plans to run in parallel.

class loom.etl.compiler.ParallelProcessGroup(plans)[source]¶

Bases: object

A group of ProcessPlan items that execute concurrently.

Parameters:: plans (tuple[ProcessPlan, ...]) – Process plans to run in parallel.

class loom.etl.compiler.StepCompilationContext(step_type, params_type, source_bindings, target_binding, config_bindings=(), agent_bindings=())[source]¶

Bases: object

Carries resolved step data required by per-step validators.

Parameters:

step_type (type[Any])
params_type (type[Any])
source_bindings (tuple[SourceBinding, ...])
target_binding (TargetBinding)
config_bindings (tuple[ConfigValueBinding, ...])
agent_bindings (tuple[AgentBinding, ...])

loom.etl.compiler.filter_plan(plan, include)[source]¶

Return a new plan keeping only the steps and processes named in include.

This is the selection run() applies for include=. A process name keeps the whole process; a step name keeps that step inside its process. A process left with no step is dropped, and a parallel group left with one member becomes that member. Execution order and the remaining parallel groups are preserved.

Parameters:

plan (PipelinePlan) – Compiled pipeline plan, as returned by compile().
include (Iterable[str]) – Class names of steps or processes to keep. A bare string is refused rather than read as its characters.

Returns:

A new PipelinePlan; plan is not modified.

Raises:

TypeError – When include is a str.
InvalidStageError – When a name in include is neither a step nor a process of plan.

Return type:

PipelinePlan

Example:

from loom.etl.compiler import ETLCompiler, filter_plan, iter_all_steps

plan = filter_plan(ETLCompiler().compile(MyPipeline), {"LoadStep"})
[step.step_type.__name__ for step in iter_all_steps(plan)]  # ["LoadStep"]

loom.etl.compiler.flatten_step_names(plan, processes=None)[source]¶

Flatten a compiled plan into the ordered tuple of step class names.

Steps are returned in their planned execution order (parallel groups are flattened in declaration order). When processes is given, only processes whose ProcessPlan.process_type.__name__ is in that tuple contribute their step names.

Parameters:

plan (PipelinePlan) – The compiled pipeline plan.
processes (tuple[str, ...] | None) – Optional whitelist of process class names. None (the default) includes every process.

Returns:

Ordered tuple of step class names ready to be passed to run via include=.

Return type:

tuple[str, …]

Example:

>>> from loom.etl.compiler import ETLCompiler, flatten_step_names
>>> plan = ETLCompiler().compile(MyPipeline)
>>> flatten_step_names(plan)
('ExtractStep', 'TransformStep', 'LoadStep')

loom.etl.compiler.iter_all_steps(plan)[source]¶

Yield every concrete step plan in a pipeline, in execution order.

Equivalent to iter_steps_in_process() over each process yielded by iter_processes().

Parameters:: plan (PipelinePlan) – Compiled pipeline plan.
Yields:: Each StepPlan of plan; a step class listed by two processes is yielded once per process.
Return type:: Iterator[StepPlan]

loom.etl.compiler.iter_processes(plan)[source]¶

Yield every concrete process plan in pipeline execution order.

Flattens top-level ParallelProcessGroup nodes while preserving the declared order within each group.

Parameters:: plan (PipelinePlan) – Compiled pipeline plan.
Yields:: Each ProcessPlan of plan, once.
Return type:: Iterator[ProcessPlan]

loom.etl.compiler.iter_steps_in_process(plan)[source]¶

Yield every concrete step plan in process execution order.

Flattens ParallelStepGroup nodes while preserving the declared order within each group.

Parameters:: plan (ProcessPlan) – Compiled process plan, e.g. one yielded by iter_processes().
Yields:: Each StepPlan of plan, once.
Return type:: Iterator[StepPlan]

loom.etl.compiler.validate_execute_signature(step_type, params_type, source_bindings, config_bindings=(), agent_bindings=())[source]¶

Validate execute() params and keyword-only source, config and agent bindings.

Parameters:

step_type (type[Any])
params_type (type[Any])
source_bindings (tuple[SourceBinding, ...])
config_bindings (tuple[ConfigValueBinding, ...])
agent_bindings (tuple[AgentBinding, ...])

Return type:

None

loom.etl.compiler.validate_param_exprs(step_type, params_type, source_bindings, target_binding)[source]¶

Raise when any ParamExpr references an undeclared params field.

A reference is declared when it is a struct field of params_type or a public property on the class — computed read-only params resolve through the same getattr chain at runtime as plain fields.

Parameters:

step_type (type[Any])
params_type (type[Any])
source_bindings (tuple[SourceBinding, ...])
target_binding (TargetBinding)

Return type:

None

loom.etl.compiler.validate_params_compat(component_type, component_params, context_params)[source]¶

Raise if component_params requires fields absent from context_params.

Parameters:

component_type (type[Any])
component_params (type[Any])
context_params (type[Any])

Return type:

None

loom.etl.compiler.validate_plan_catalog(plan, catalog)[source]¶

Walk the plan validating all table references against the catalog.

Parameters:

plan (PipelinePlan)
catalog (TableDiscovery)

Return type:

None

loom.etl.compiler.validate_plan_temps(plan)[source]¶

Walk the plan validating temp forward-references and name uniqueness.

Parameters:: plan (PipelinePlan)
Return type:: None

loom.etl.compiler.validate_process_catalog(plan, catalog)[source]¶

Validate all steps in a process against the catalog.

Parameters:

plan (ProcessPlan)
catalog (TableDiscovery)

Return type:

None

loom.etl.compiler.validate_step(ctx)[source]¶

Run all per-step compile-time validators against ctx.

Parameters:: ctx (StepCompilationContext)
Return type:: None

loom.etl.compiler.validate_step_catalog(plan, catalog)[source]¶

Validate a single step’s sources and target against the catalog.

Parameters:

plan (StepPlan)
catalog (TableDiscovery)

Return type:

None

loom.etl.compiler.validate_upsert_spec(step_type, spec)[source]¶

Validate UPSERT/UPDATE merge constraints at compile time.

UpdateSpec mirrors the upsert merge fields, so the same checks apply — only the builder method named in the error changes.

Parameters:

step_type (type[Any])
spec (AppendSpec | ReplaceSpec | ReplacePartitionsSpec | ReplaceWhereSpec | UpsertSpec | UpdateSpec | HistorifySpec | FileSpec | TempSpec | TempFanInSpec | ClientSpec)

Return type:

None