nirs4all.pipeline.storage.chain_builder module

Chain builder for converting ExecutionTrace to WorkspaceStore chain format.

Bridges the trace/artifact system and the DuckDB WorkspaceStore by converting an ExecutionTrace (recorded during pipeline execution) into the chain dict format expected by store.save_chain().

When a pipeline has multiple model steps, the builder produces one chain per model step so that each prediction can be resolved independently.

Example

>>> from nirs4all.pipeline.storage.chain_builder import ChainBuilder
>>> chain_builder = ChainBuilder(trace, artifact_registry)
>>> for chain_data in chain_builder.build_all():
...     store.save_chain(pipeline_id=pipeline_id, **chain_data)

class nirs4all.pipeline.storage.chain_builder.ChainBuilder(trace: ExecutionTrace, artifact_registry: Any = None)

Bases: object

Converts an ExecutionTrace into chain dicts for WorkspaceStore.save_chain().

The builder extracts the ordered sequence of non-skipped steps, identifies model steps, collects fold and shared artifact IDs, and produces chain descriptors ready for DuckDB persistence.

When the trace has multiple model steps, build_all() produces one chain per model. build() returns the chain for the trace’s primary model_step_index (backward-compatible).

Parameters:

• trace – Finalized ExecutionTrace from TraceRecorder.

• artifact_registry – ArtifactRegistry that holds artifact records produced during this pipeline execution.

Example

>>> builder = ChainBuilder(trace, artifact_registry)
>>> for chain_data in builder.build_all():
...     store.save_chain(pipeline_id=pid, **chain_data)

build() → Dict[str, Any]

Build one chain dict for the trace’s primary model step.

Returns:

Dictionary with keys matching WorkspaceStore.save_chain() parameters.

build_all() → List[Dict[str, Any]]

Build one chain dict per model step found in the trace.

Returns:

List of chain dicts, one per model step.