Directed acyclic graph execution framework for deterministic state propagation.
dagpipe executes pipelines composed of nodes connected in a directed acyclic graph (DAG). Each node receives an immutable State and optionally produces derived states for downstream nodes.
Base class for nodes whose execution is asynchronous.
Subclasses implement resolve_async (an async generator yielding derived State objects). The engine dispatches to resolve_async when running an async traversal (see Engine.run_async).
Sync-only engines (and the base Node.run) treat an AsyncNode as a no-op consumer: calling run on an AsyncNode returns no states, signalling that an async engine is required.
Execution engine responsible for running pipeline logic.
Notes
Responsibilities:
- Accepts either a linear sequence of `Node` objects or a `Graph`\n defining execution topology.\n- Propagates immutable `State` objects through `Node` objects and\n collects terminal states.\n- Supports synchronous (`run`) and asynchronous (`run_async`)\n execution, dispatching per-node.\n- Supports step-wise / resumable execution and progress hooks.\n
Guarantees:
- Never mutates `State`, `Node`, or `Graph` instances.\n- `State` objects are never modified in place; each branch produces\n independent instances.\n- Execution order is deterministic and follows graph or pipeline topology.\n- Thread-safe for concurrent execution.\n
Execute the pipeline step-by-step, yielding one StepResult per step.
Parameters:
Name Type Description Default rootState
Initial execution state.
required resume_fromint | None
Skip steps at index < resume_from (for resume-after-partial). Steps are 0-indexed.
Noneon_stepStepHook | None
Callback (step, status, message) invoked per step; falls back to the engine-level hook when unset.
None
Yields:
Name Type Description StepResultStepResult
One per executed node/step, carrying the produced states.
Notes
This is a synchronous, generator-based checkpoint interface compatible with the imperative resume-by-step behaviour of the legacy orchestrator. Use run_steps_async for async nodes.
Directed Acyclic Graph defining execution topology of Node objects.
Notes
Responsibilities:
- Stores node connectivity and validates that the topology remains acyclic.\n- Structure determines how `State` flows between nodes during execution.\n
Guarantees:
- Topology is acyclic. Node relationships remain consistent.\n- Thread-safe for concurrent reads after construction.\n
Create an empty Graph.
"},{"location":"#dagpipe.Graph--initializes-node-registry-and-edge-mappings","title":"Initializes node registry and edge mappings.","text":""},{"location":"#dagpipe.Graph-functions","title":"Functions","text":""},{"location":"#dagpipe.Graph.__repr__","title":"__repr__","text":"
Unique identifier of the node (snake_case dotted format).
namestr
Human-readable display name.
Notes
Responsibilities:
- Represents a deterministic unit of execution in the pipeline graph.\n- Consumes one `State` and produces zero, one, or many derived states.\n- Defines execution logic and enables branching, filtering, and transformation.\n
Guarantees:
- Nodes must never mutate the input `State`.\n- Instances are singletons per subclass and reused across executions.\n
Stateless subclasses (no parameterized __init__) share one singleton instance per class \u2014 matching the original dagpipe behaviour underpinning set_registry-style configuration. Subclasses that declare an __init__ requiring instance-state arguments get a fresh instance per construction so pipeline builders can inject per-run dependencies.
- Subclasses implement specific resolution behavior.\n- Must not mutate input state.\n- Should use `fork()` to create child states.\n- May yield zero states to terminate a branch.\n
Immutable hierarchical container with dot-path access.
Attributes:
Name Type Description _dataMapping[str, Any]
Immutable hierarchical data structure.
Notes
Responsibilities:
- Stores execution data used by `State`.\n- Supports efficient atomic updates without modifying existing instances.\n- `Payload` instances are fully thread-safe due to immutability.\n
Executable pipeline created from YAML configuration.
Attributes:
Name Type Description engineEngine
Execution engine responsible for running the pipeline.
state_clsType[State]
Dynamically created State subclass with configured schema.
initial_payloadPayload
Default payload used when execution begins.
Notes
Responsibilities:
- Encapsulates engine, state type, and initial payload.\n- Provides a simplified interface for executing configured pipelines.\n- Safe for concurrent execution if underlying nodes are thread-safe.\n
Lightweight progress payload emitted by engine step hooks.
Mirrors the imperative ProgressMessage used by the legacy orchestrator so callers can surface counts/lines/errors without coupling the engine to pydantic.
Immutable execution state propagated through dagpipe pipeline.
Attributes:
Name Type Description payloadPayload
Execution data container.
schemaClassVar[Schema]
Payload validation schema.
confidencefloat
Execution confidence score.
parentOptional[State]
Parent state reference.
depthint
Execution depth.
historyTuple[str, ...]
Ordered node execution lineage.
metadataDict[str, Any]
Execution metadata.
Notes
Responsibilities:
- Represents a complete execution snapshot at a specific point in\n pipeline traversal.\n- Fundamental unit of execution in `dagpipe`.\n- Fully thread-safe due to immutability.\n
Name Type Description Default payload_updateMapping[str, Any] | None
Dot-path updates applied to the payload.
Noneconfidence_deltafloat
Adjustment applied to current confidence.
0.0node_idstr | None
Identifier of the node creating this state.
Nonemetadata_updateMapping[str, Any] | None
Updates merged into state metadata.
None
Returns:
Name Type Description StateState
A new immutable State instance.
Notes
Guarantees:
- This is the only supported mechanism for modifying execution data.\n- Validates payload updates, preserves lineage, increments depth,\n and appends to history.\n
Execution engine responsible for running pipelines and graphs.
The Engine executes Node objects and propagates immutable State instances through either a linear sequence or a directed acyclic graph (Graph). It orchestrates execution order, branching, and state propagation.
Execution engine responsible for running pipeline logic.
Notes
Responsibilities:
- Accepts either a linear sequence of `Node` objects or a `Graph`\n defining execution topology.\n- Propagates immutable `State` objects through `Node` objects and\n collects terminal states.\n- Supports synchronous (`run`) and asynchronous (`run_async`)\n execution, dispatching per-node.\n- Supports step-wise / resumable execution and progress hooks.\n
Guarantees:
- Never mutates `State`, `Node`, or `Graph` instances.\n- `State` objects are never modified in place; each branch produces\n independent instances.\n- Execution order is deterministic and follows graph or pipeline topology.\n- Thread-safe for concurrent execution.\n
Execute the pipeline step-by-step, yielding one StepResult per step.
Parameters:
Name Type Description Default rootState
Initial execution state.
required resume_fromint | None
Skip steps at index < resume_from (for resume-after-partial). Steps are 0-indexed.
Noneon_stepStepHook | None
Callback (step, status, message) invoked per step; falls back to the engine-level hook when unset.
None
Yields:
Name Type Description StepResultStepResult
One per executed node/step, carrying the produced states.
Notes
This is a synchronous, generator-based checkpoint interface compatible with the imperative resume-by-step behaviour of the legacy orchestrator. Use run_steps_async for async nodes.
Lightweight progress payload emitted by engine step hooks.
Mirrors the imperative ProgressMessage used by the legacy orchestrator so callers can surface counts/lines/errors without coupling the engine to pydantic.
Directed Acyclic Graph defining execution topology of Node objects.
Notes
Responsibilities:
- Stores node connectivity and validates that the topology remains acyclic.\n- Structure determines how `State` flows between nodes during execution.\n
Guarantees:
- Topology is acyclic. Node relationships remain consistent.\n- Thread-safe for concurrent reads after construction.\n
Create an empty Graph.
"},{"location":"graph/#dagpipe.graph.Graph--initializes-node-registry-and-edge-mappings","title":"Initializes node registry and edge mappings.","text":""},{"location":"graph/#dagpipe.graph.Graph-functions","title":"Functions","text":""},{"location":"graph/#dagpipe.graph.Graph.__repr__","title":"__repr__","text":"
Base class for nodes whose execution is asynchronous.
Subclasses implement resolve_async (an async generator yielding derived State objects). The engine dispatches to resolve_async when running an async traversal (see Engine.run_async).
Sync-only engines (and the base Node.run) treat an AsyncNode as a no-op consumer: calling run on an AsyncNode returns no states, signalling that an async engine is required.
Unique identifier of the node (snake_case dotted format).
namestr
Human-readable display name.
Notes
Responsibilities:
- Represents a deterministic unit of execution in the pipeline graph.\n- Consumes one `State` and produces zero, one, or many derived states.\n- Defines execution logic and enables branching, filtering, and transformation.\n
Guarantees:
- Nodes must never mutate the input `State`.\n- Instances are singletons per subclass and reused across executions.\n
Stateless subclasses (no parameterized __init__) share one singleton instance per class \u2014 matching the original dagpipe behaviour underpinning set_registry-style configuration. Subclasses that declare an __init__ requiring instance-state arguments get a fresh instance per construction so pipeline builders can inject per-run dependencies.
- Subclasses implement specific resolution behavior.\n- Must not mutate input state.\n- Should use `fork()` to create child states.\n- May yield zero states to terminate a branch.\n
The State represents a single point in pipeline execution. It contains arbitrary data and metadata and is designed to be immutable. Instead of modifying an existing state, nodes create new child states via fork().
Immutable hierarchical container with dot-path access.
Attributes:
Name Type Description _dataMapping[str, Any]
Immutable hierarchical data structure.
Notes
Responsibilities:
- Stores execution data used by `State`.\n- Supports efficient atomic updates without modifying existing instances.\n- `Payload` instances are fully thread-safe due to immutability.\n
Immutable execution state propagated through dagpipe pipeline.
Attributes:
Name Type Description payloadPayload
Execution data container.
schemaClassVar[Schema]
Payload validation schema.
confidencefloat
Execution confidence score.
parentOptional[State]
Parent state reference.
depthint
Execution depth.
historyTuple[str, ...]
Ordered node execution lineage.
metadataDict[str, Any]
Execution metadata.
Notes
Responsibilities:
- Represents a complete execution snapshot at a specific point in\n pipeline traversal.\n- Fundamental unit of execution in `dagpipe`.\n- Fully thread-safe due to immutability.\n
Name Type Description Default payload_updateMapping[str, Any] | None
Dot-path updates applied to the payload.
Noneconfidence_deltafloat
Adjustment applied to current confidence.
0.0node_idstr | None
Identifier of the node creating this state.
Nonemetadata_updateMapping[str, Any] | None
Updates merged into state metadata.
None
Returns:
Name Type Description StateState
A new immutable State instance.
Notes
Guarantees:
- This is the only supported mechanism for modifying execution data.\n- Validates payload updates, preserves lineage, increments depth,\n and appends to history.\n
Creates fully configured pipeline objects from declarative YAML definitions, including Schema, State subclasses, Node instances, Graph topology, and initial payloads.
Executable pipeline created from YAML configuration.
Attributes:
Name Type Description engineEngine
Execution engine responsible for running the pipeline.
state_clsType[State]
Dynamically created State subclass with configured schema.
initial_payloadPayload
Default payload used when execution begins.
Notes
Responsibilities:
- Encapsulates engine, state type, and initial payload.\n- Provides a simplified interface for executing configured pipelines.\n- Safe for concurrent execution if underlying nodes are thread-safe.\n
Directed acyclic graph execution framework for deterministic state propagation.
dagpipe executes pipelines composed of nodes connected in a directed acyclic graph (DAG). Each node receives an immutable State and optionally produces derived states for downstream nodes.
Base class for nodes whose execution is asynchronous.
Subclasses implement resolve_async (an async generator yielding derived State objects). The engine dispatches to resolve_async when running an async traversal (see Engine.run_async).
Sync-only engines (and the base Node.run) treat an AsyncNode as a no-op consumer: calling run on an AsyncNode returns no states, signalling that an async engine is required.
Execution engine responsible for running pipeline logic.
Notes
Responsibilities:
- Accepts either a linear sequence of `Node` objects or a `Graph`\n defining execution topology.\n- Propagates immutable `State` objects through `Node` objects and\n collects terminal states.\n- Supports synchronous (`run`) and asynchronous (`run_async`)\n execution, dispatching per-node.\n- Supports step-wise / resumable execution and progress hooks.\n
Guarantees:
- Never mutates `State`, `Node`, or `Graph` instances.\n- `State` objects are never modified in place; each branch produces\n independent instances.\n- Execution order is deterministic and follows graph or pipeline topology.\n- Thread-safe for concurrent execution.\n
Execute the pipeline step-by-step, yielding one StepResult per step.
Parameters:
Name Type Description Default rootState
Initial execution state.
required resume_fromint | None
Skip steps at index < resume_from (for resume-after-partial). Steps are 0-indexed.
Noneon_stepStepHook | None
Callback (step, status, message) invoked per step; falls back to the engine-level hook when unset.
None
Yields:
Name Type Description StepResultStepResult
One per executed node/step, carrying the produced states.
Notes
This is a synchronous, generator-based checkpoint interface compatible with the imperative resume-by-step behaviour of the legacy orchestrator. Use run_steps_async for async nodes.
Directed Acyclic Graph defining execution topology of Node objects.
Notes
Responsibilities:
- Stores node connectivity and validates that the topology remains acyclic.\n- Structure determines how `State` flows between nodes during execution.\n
Guarantees:
- Topology is acyclic. Node relationships remain consistent.\n- Thread-safe for concurrent reads after construction.\n
Create an empty Graph.
"},{"location":"dagpipe/#dagpipe.Graph--initializes-node-registry-and-edge-mappings","title":"Initializes node registry and edge mappings.","text":""},{"location":"dagpipe/#dagpipe.Graph-functions","title":"Functions","text":""},{"location":"dagpipe/#dagpipe.Graph.__repr__","title":"__repr__","text":"
Unique identifier of the node (snake_case dotted format).
namestr
Human-readable display name.
Notes
Responsibilities:
- Represents a deterministic unit of execution in the pipeline graph.\n- Consumes one `State` and produces zero, one, or many derived states.\n- Defines execution logic and enables branching, filtering, and transformation.\n
Guarantees:
- Nodes must never mutate the input `State`.\n- Instances are singletons per subclass and reused across executions.\n
Stateless subclasses (no parameterized __init__) share one singleton instance per class \u2014 matching the original dagpipe behaviour underpinning set_registry-style configuration. Subclasses that declare an __init__ requiring instance-state arguments get a fresh instance per construction so pipeline builders can inject per-run dependencies.
- Subclasses implement specific resolution behavior.\n- Must not mutate input state.\n- Should use `fork()` to create child states.\n- May yield zero states to terminate a branch.\n
Immutable hierarchical container with dot-path access.
Attributes:
Name Type Description _dataMapping[str, Any]
Immutable hierarchical data structure.
Notes
Responsibilities:
- Stores execution data used by `State`.\n- Supports efficient atomic updates without modifying existing instances.\n- `Payload` instances are fully thread-safe due to immutability.\n
Executable pipeline created from YAML configuration.
Attributes:
Name Type Description engineEngine
Execution engine responsible for running the pipeline.
state_clsType[State]
Dynamically created State subclass with configured schema.
initial_payloadPayload
Default payload used when execution begins.
Notes
Responsibilities:
- Encapsulates engine, state type, and initial payload.\n- Provides a simplified interface for executing configured pipelines.\n- Safe for concurrent execution if underlying nodes are thread-safe.\n
Lightweight progress payload emitted by engine step hooks.
Mirrors the imperative ProgressMessage used by the legacy orchestrator so callers can surface counts/lines/errors without coupling the engine to pydantic.
Immutable execution state propagated through dagpipe pipeline.
Attributes:
Name Type Description payloadPayload
Execution data container.
schemaClassVar[Schema]
Payload validation schema.
confidencefloat
Execution confidence score.
parentOptional[State]
Parent state reference.
depthint
Execution depth.
historyTuple[str, ...]
Ordered node execution lineage.
metadataDict[str, Any]
Execution metadata.
Notes
Responsibilities:
- Represents a complete execution snapshot at a specific point in\n pipeline traversal.\n- Fundamental unit of execution in `dagpipe`.\n- Fully thread-safe due to immutability.\n
Name Type Description Default payload_updateMapping[str, Any] | None
Dot-path updates applied to the payload.
Noneconfidence_deltafloat
Adjustment applied to current confidence.
0.0node_idstr | None
Identifier of the node creating this state.
Nonemetadata_updateMapping[str, Any] | None
Updates merged into state metadata.
None
Returns:
Name Type Description StateState
A new immutable State instance.
Notes
Guarantees:
- This is the only supported mechanism for modifying execution data.\n- Validates payload updates, preserves lineage, increments depth,\n and appends to history.\n
Execution engine responsible for running pipelines and graphs.
The Engine executes Node objects and propagates immutable State instances through either a linear sequence or a directed acyclic graph (Graph). It orchestrates execution order, branching, and state propagation.
Execution engine responsible for running pipeline logic.
Notes
Responsibilities:
- Accepts either a linear sequence of `Node` objects or a `Graph`\n defining execution topology.\n- Propagates immutable `State` objects through `Node` objects and\n collects terminal states.\n- Supports synchronous (`run`) and asynchronous (`run_async`)\n execution, dispatching per-node.\n- Supports step-wise / resumable execution and progress hooks.\n
Guarantees:
- Never mutates `State`, `Node`, or `Graph` instances.\n- `State` objects are never modified in place; each branch produces\n independent instances.\n- Execution order is deterministic and follows graph or pipeline topology.\n- Thread-safe for concurrent execution.\n
Execute the pipeline step-by-step, yielding one StepResult per step.
Parameters:
Name Type Description Default rootState
Initial execution state.
required resume_fromint | None
Skip steps at index < resume_from (for resume-after-partial). Steps are 0-indexed.
Noneon_stepStepHook | None
Callback (step, status, message) invoked per step; falls back to the engine-level hook when unset.
None
Yields:
Name Type Description StepResultStepResult
One per executed node/step, carrying the produced states.
Notes
This is a synchronous, generator-based checkpoint interface compatible with the imperative resume-by-step behaviour of the legacy orchestrator. Use run_steps_async for async nodes.
Lightweight progress payload emitted by engine step hooks.
Mirrors the imperative ProgressMessage used by the legacy orchestrator so callers can surface counts/lines/errors without coupling the engine to pydantic.
Directed Acyclic Graph defining execution topology of Node objects.
Notes
Responsibilities:
- Stores node connectivity and validates that the topology remains acyclic.\n- Structure determines how `State` flows between nodes during execution.\n
Guarantees:
- Topology is acyclic. Node relationships remain consistent.\n- Thread-safe for concurrent reads after construction.\n
Create an empty Graph.
"},{"location":"dagpipe/graph/#dagpipe.graph.Graph--initializes-node-registry-and-edge-mappings","title":"Initializes node registry and edge mappings.","text":""},{"location":"dagpipe/graph/#dagpipe.graph.Graph-functions","title":"Functions","text":""},{"location":"dagpipe/graph/#dagpipe.graph.Graph.__repr__","title":"__repr__","text":"
Base class for nodes whose execution is asynchronous.
Subclasses implement resolve_async (an async generator yielding derived State objects). The engine dispatches to resolve_async when running an async traversal (see Engine.run_async).
Sync-only engines (and the base Node.run) treat an AsyncNode as a no-op consumer: calling run on an AsyncNode returns no states, signalling that an async engine is required.
Unique identifier of the node (snake_case dotted format).
namestr
Human-readable display name.
Notes
Responsibilities:
- Represents a deterministic unit of execution in the pipeline graph.\n- Consumes one `State` and produces zero, one, or many derived states.\n- Defines execution logic and enables branching, filtering, and transformation.\n
Guarantees:
- Nodes must never mutate the input `State`.\n- Instances are singletons per subclass and reused across executions.\n
Stateless subclasses (no parameterized __init__) share one singleton instance per class \u2014 matching the original dagpipe behaviour underpinning set_registry-style configuration. Subclasses that declare an __init__ requiring instance-state arguments get a fresh instance per construction so pipeline builders can inject per-run dependencies.
- Subclasses implement specific resolution behavior.\n- Must not mutate input state.\n- Should use `fork()` to create child states.\n- May yield zero states to terminate a branch.\n
The State represents a single point in pipeline execution. It contains arbitrary data and metadata and is designed to be immutable. Instead of modifying an existing state, nodes create new child states via fork().
Immutable hierarchical container with dot-path access.
Attributes:
Name Type Description _dataMapping[str, Any]
Immutable hierarchical data structure.
Notes
Responsibilities:
- Stores execution data used by `State`.\n- Supports efficient atomic updates without modifying existing instances.\n- `Payload` instances are fully thread-safe due to immutability.\n
Immutable execution state propagated through dagpipe pipeline.
Attributes:
Name Type Description payloadPayload
Execution data container.
schemaClassVar[Schema]
Payload validation schema.
confidencefloat
Execution confidence score.
parentOptional[State]
Parent state reference.
depthint
Execution depth.
historyTuple[str, ...]
Ordered node execution lineage.
metadataDict[str, Any]
Execution metadata.
Notes
Responsibilities:
- Represents a complete execution snapshot at a specific point in\n pipeline traversal.\n- Fundamental unit of execution in `dagpipe`.\n- Fully thread-safe due to immutability.\n
Name Type Description Default payload_updateMapping[str, Any] | None
Dot-path updates applied to the payload.
Noneconfidence_deltafloat
Adjustment applied to current confidence.
0.0node_idstr | None
Identifier of the node creating this state.
Nonemetadata_updateMapping[str, Any] | None
Updates merged into state metadata.
None
Returns:
Name Type Description StateState
A new immutable State instance.
Notes
Guarantees:
- This is the only supported mechanism for modifying execution data.\n- Validates payload updates, preserves lineage, increments depth,\n and appends to history.\n
Creates fully configured pipeline objects from declarative YAML definitions, including Schema, State subclasses, Node instances, Graph topology, and initial payloads.
Executable pipeline created from YAML configuration.
Attributes:
Name Type Description engineEngine
Execution engine responsible for running the pipeline.
state_clsType[State]
Dynamically created State subclass with configured schema.
initial_payloadPayload
Default payload used when execution begins.
Notes
Responsibilities:
- Encapsulates engine, state type, and initial payload.\n- Provides a simplified interface for executing configured pipelines.\n- Safe for concurrent execution if underlying nodes are thread-safe.\n
- Loads YAML configuration and builds schema.\n- Creates `State` subclass and loads `Node` instances.\n- Builds `Graph` topology and initializes `Engine`.\n
"}]}
\ No newline at end of file
+{"config":{"lang":["en"],"separator":"[\\s\\-]+","pipeline":["stopWordFilter"]},"docs":[{"location":"","title":"dagpipe","text":""},{"location":"#dagpipe","title":"dagpipe","text":""},{"location":"#dagpipe--summary","title":"Summary","text":"
Directed acyclic graph execution framework for deterministic state propagation.
dagpipe executes pipelines composed of nodes connected in a directed acyclic graph (DAG). Each node receives an immutable State and optionally produces derived states for downstream nodes.
Base class for nodes whose execution is asynchronous.
Subclasses implement resolve_async (an async generator yielding derived State objects). The engine dispatches to resolve_async when running an async traversal (see Engine.run_async).
Sync-only engines (and the base Node.run) treat an AsyncNode as a no-op consumer: calling run on an AsyncNode returns no states, signalling that an async engine is required.
Execution engine responsible for running pipeline logic.
Notes
Responsibilities:
- Accepts either a linear sequence of `Node` objects or a `Graph`\n defining execution topology.\n- Propagates immutable `State` objects through `Node` objects and\n collects terminal states.\n- Supports synchronous (`run`) and asynchronous (`run_async`)\n execution, dispatching per-node.\n- Supports step-wise / resumable execution and progress hooks.\n
Guarantees:
- Never mutates `State`, `Node`, or `Graph` instances.\n- `State` objects are never modified in place; each branch produces\n independent instances.\n- Execution order is deterministic and follows graph or pipeline topology.\n- Thread-safe for concurrent execution.\n
Execute the pipeline step-by-step, yielding one StepResult per step.
Parameters:
Name Type Description Default rootState
Initial execution state.
required resume_fromint | None
Skip steps at index < resume_from (for resume-after-partial). Steps are 0-indexed.
Noneon_stepStepHook | None
Callback (step, status, message) invoked per step; falls back to the engine-level hook when unset.
None
Yields:
Name Type Description StepResultStepResult
One per executed node/step, carrying the produced states.
Notes
This is a synchronous, generator-based checkpoint interface compatible with the imperative resume-by-step behaviour of the legacy orchestrator. Use run_steps_async for async nodes.
Directed Acyclic Graph defining execution topology of Node objects.
Notes
Responsibilities:
- Stores node connectivity and validates that the topology remains acyclic.\n- Structure determines how `State` flows between nodes during execution.\n
Guarantees:
- Topology is acyclic. Node relationships remain consistent.\n- Thread-safe for concurrent reads after construction.\n
Create an empty Graph.
"},{"location":"#dagpipe.Graph--initializes-node-registry-and-edge-mappings","title":"Initializes node registry and edge mappings.","text":""},{"location":"#dagpipe.Graph-functions","title":"Functions","text":""},{"location":"#dagpipe.Graph.__repr__","title":"__repr__","text":"
Unique identifier of the node (snake_case dotted format).
namestr
Human-readable display name.
Notes
Responsibilities:
- Represents a deterministic unit of execution in the pipeline graph.\n- Consumes one `State` and produces zero, one, or many derived states.\n- Defines execution logic and enables branching, filtering, and transformation.\n
Guarantees:
- Nodes must never mutate the input `State`.\n- Instances are singletons per subclass and reused across executions.\n
Stateless subclasses (no parameterized __init__) share one singleton instance per class \u2014 matching the original dagpipe behaviour underpinning set_registry-style configuration. Subclasses that declare an __init__ requiring instance-state arguments get a fresh instance per construction so pipeline builders can inject per-run dependencies.
- Subclasses implement specific resolution behavior.\n- Must not mutate input state.\n- Should use `fork()` to create child states.\n- May yield zero states to terminate a branch.\n
Immutable hierarchical container with dot-path access.
Attributes:
Name Type Description _dataMapping[str, Any]
Immutable hierarchical data structure.
Notes
Responsibilities:
- Stores execution data used by `State`.\n- Supports efficient atomic updates without modifying existing instances.\n- `Payload` instances are fully thread-safe due to immutability.\n
Executable pipeline created from YAML configuration.
Attributes:
Name Type Description engineEngine
Execution engine responsible for running the pipeline.
state_clsType[State]
Dynamically created State subclass with configured schema.
initial_payloadPayload
Default payload used when execution begins.
Notes
Responsibilities:
- Encapsulates engine, state type, and initial payload.\n- Provides a simplified interface for executing configured pipelines.\n- Safe for concurrent execution if underlying nodes are thread-safe.\n
Lightweight progress payload emitted by engine step hooks.
Mirrors the imperative ProgressMessage used by the legacy orchestrator so callers can surface counts/lines/errors without coupling the engine to pydantic.
Immutable execution state propagated through dagpipe pipeline.
Attributes:
Name Type Description payloadPayload
Execution data container.
schemaClassVar[Schema]
Payload validation schema.
confidencefloat
Execution confidence score.
parentOptional[State]
Parent state reference.
depthint
Execution depth.
historyTuple[str, ...]
Ordered node execution lineage.
metadataDict[str, Any]
Execution metadata.
Notes
Responsibilities:
- Represents a complete execution snapshot at a specific point in\n pipeline traversal.\n- Fundamental unit of execution in `dagpipe`.\n- Fully thread-safe due to immutability.\n
Name Type Description Default payload_updateMapping[str, Any] | None
Dot-path updates applied to the payload.
Noneconfidence_deltafloat
Adjustment applied to current confidence.
0.0node_idstr | None
Identifier of the node creating this state.
Nonemetadata_updateMapping[str, Any] | None
Updates merged into state metadata.
None
Returns:
Name Type Description StateState
A new immutable State instance.
Notes
Guarantees:
- This is the only supported mechanism for modifying execution data.\n- Validates payload updates, preserves lineage, increments depth,\n and appends to history.\n
Execution engine responsible for running pipelines and graphs.
The Engine executes Node objects and propagates immutable State instances through either a linear sequence or a directed acyclic graph (Graph). It orchestrates execution order, branching, and state propagation.
Execution engine responsible for running pipeline logic.
Notes
Responsibilities:
- Accepts either a linear sequence of `Node` objects or a `Graph`\n defining execution topology.\n- Propagates immutable `State` objects through `Node` objects and\n collects terminal states.\n- Supports synchronous (`run`) and asynchronous (`run_async`)\n execution, dispatching per-node.\n- Supports step-wise / resumable execution and progress hooks.\n
Guarantees:
- Never mutates `State`, `Node`, or `Graph` instances.\n- `State` objects are never modified in place; each branch produces\n independent instances.\n- Execution order is deterministic and follows graph or pipeline topology.\n- Thread-safe for concurrent execution.\n
Execute the pipeline step-by-step, yielding one StepResult per step.
Parameters:
Name Type Description Default rootState
Initial execution state.
required resume_fromint | None
Skip steps at index < resume_from (for resume-after-partial). Steps are 0-indexed.
Noneon_stepStepHook | None
Callback (step, status, message) invoked per step; falls back to the engine-level hook when unset.
None
Yields:
Name Type Description StepResultStepResult
One per executed node/step, carrying the produced states.
Notes
This is a synchronous, generator-based checkpoint interface compatible with the imperative resume-by-step behaviour of the legacy orchestrator. Use run_steps_async for async nodes.
Lightweight progress payload emitted by engine step hooks.
Mirrors the imperative ProgressMessage used by the legacy orchestrator so callers can surface counts/lines/errors without coupling the engine to pydantic.
Directed Acyclic Graph defining execution topology of Node objects.
Notes
Responsibilities:
- Stores node connectivity and validates that the topology remains acyclic.\n- Structure determines how `State` flows between nodes during execution.\n
Guarantees:
- Topology is acyclic. Node relationships remain consistent.\n- Thread-safe for concurrent reads after construction.\n
Create an empty Graph.
"},{"location":"graph/#dagpipe.graph.Graph--initializes-node-registry-and-edge-mappings","title":"Initializes node registry and edge mappings.","text":""},{"location":"graph/#dagpipe.graph.Graph-functions","title":"Functions","text":""},{"location":"graph/#dagpipe.graph.Graph.__repr__","title":"__repr__","text":"
Base class for nodes whose execution is asynchronous.
Subclasses implement resolve_async (an async generator yielding derived State objects). The engine dispatches to resolve_async when running an async traversal (see Engine.run_async).
Sync-only engines (and the base Node.run) treat an AsyncNode as a no-op consumer: calling run on an AsyncNode returns no states, signalling that an async engine is required.
Unique identifier of the node (snake_case dotted format).
namestr
Human-readable display name.
Notes
Responsibilities:
- Represents a deterministic unit of execution in the pipeline graph.\n- Consumes one `State` and produces zero, one, or many derived states.\n- Defines execution logic and enables branching, filtering, and transformation.\n
Guarantees:
- Nodes must never mutate the input `State`.\n- Instances are singletons per subclass and reused across executions.\n
Stateless subclasses (no parameterized __init__) share one singleton instance per class \u2014 matching the original dagpipe behaviour underpinning set_registry-style configuration. Subclasses that declare an __init__ requiring instance-state arguments get a fresh instance per construction so pipeline builders can inject per-run dependencies.
- Subclasses implement specific resolution behavior.\n- Must not mutate input state.\n- Should use `fork()` to create child states.\n- May yield zero states to terminate a branch.\n
The State represents a single point in pipeline execution. It contains arbitrary data and metadata and is designed to be immutable. Instead of modifying an existing state, nodes create new child states via fork().
Immutable hierarchical container with dot-path access.
Attributes:
Name Type Description _dataMapping[str, Any]
Immutable hierarchical data structure.
Notes
Responsibilities:
- Stores execution data used by `State`.\n- Supports efficient atomic updates without modifying existing instances.\n- `Payload` instances are fully thread-safe due to immutability.\n
Immutable execution state propagated through dagpipe pipeline.
Attributes:
Name Type Description payloadPayload
Execution data container.
schemaClassVar[Schema]
Payload validation schema.
confidencefloat
Execution confidence score.
parentOptional[State]
Parent state reference.
depthint
Execution depth.
historyTuple[str, ...]
Ordered node execution lineage.
metadataDict[str, Any]
Execution metadata.
Notes
Responsibilities:
- Represents a complete execution snapshot at a specific point in\n pipeline traversal.\n- Fundamental unit of execution in `dagpipe`.\n- Fully thread-safe due to immutability.\n
Name Type Description Default payload_updateMapping[str, Any] | None
Dot-path updates applied to the payload.
Noneconfidence_deltafloat
Adjustment applied to current confidence.
0.0node_idstr | None
Identifier of the node creating this state.
Nonemetadata_updateMapping[str, Any] | None
Updates merged into state metadata.
None
Returns:
Name Type Description StateState
A new immutable State instance.
Notes
Guarantees:
- This is the only supported mechanism for modifying execution data.\n- Validates payload updates, preserves lineage, increments depth,\n and appends to history.\n
Creates fully configured pipeline objects from declarative YAML definitions, including Schema, State subclasses, Node instances, Graph topology, and initial payloads.
Executable pipeline created from YAML configuration.
Attributes:
Name Type Description engineEngine
Execution engine responsible for running the pipeline.
state_clsType[State]
Dynamically created State subclass with configured schema.
initial_payloadPayload
Default payload used when execution begins.
Notes
Responsibilities:
- Encapsulates engine, state type, and initial payload.\n- Provides a simplified interface for executing configured pipelines.\n- Safe for concurrent execution if underlying nodes are thread-safe.\n
Directed acyclic graph execution framework for deterministic state propagation.
dagpipe executes pipelines composed of nodes connected in a directed acyclic graph (DAG). Each node receives an immutable State and optionally produces derived states for downstream nodes.
Base class for nodes whose execution is asynchronous.
Subclasses implement resolve_async (an async generator yielding derived State objects). The engine dispatches to resolve_async when running an async traversal (see Engine.run_async).
Sync-only engines (and the base Node.run) treat an AsyncNode as a no-op consumer: calling run on an AsyncNode returns no states, signalling that an async engine is required.
Execution engine responsible for running pipeline logic.
Notes
Responsibilities:
- Accepts either a linear sequence of `Node` objects or a `Graph`\n defining execution topology.\n- Propagates immutable `State` objects through `Node` objects and\n collects terminal states.\n- Supports synchronous (`run`) and asynchronous (`run_async`)\n execution, dispatching per-node.\n- Supports step-wise / resumable execution and progress hooks.\n
Guarantees:
- Never mutates `State`, `Node`, or `Graph` instances.\n- `State` objects are never modified in place; each branch produces\n independent instances.\n- Execution order is deterministic and follows graph or pipeline topology.\n- Thread-safe for concurrent execution.\n
Execute the pipeline step-by-step, yielding one StepResult per step.
Parameters:
Name Type Description Default rootState
Initial execution state.
required resume_fromint | None
Skip steps at index < resume_from (for resume-after-partial). Steps are 0-indexed.
Noneon_stepStepHook | None
Callback (step, status, message) invoked per step; falls back to the engine-level hook when unset.
None
Yields:
Name Type Description StepResultStepResult
One per executed node/step, carrying the produced states.
Notes
This is a synchronous, generator-based checkpoint interface compatible with the imperative resume-by-step behaviour of the legacy orchestrator. Use run_steps_async for async nodes.
Directed Acyclic Graph defining execution topology of Node objects.
Notes
Responsibilities:
- Stores node connectivity and validates that the topology remains acyclic.\n- Structure determines how `State` flows between nodes during execution.\n
Guarantees:
- Topology is acyclic. Node relationships remain consistent.\n- Thread-safe for concurrent reads after construction.\n
Create an empty Graph.
"},{"location":"dagpipe/#dagpipe.Graph--initializes-node-registry-and-edge-mappings","title":"Initializes node registry and edge mappings.","text":""},{"location":"dagpipe/#dagpipe.Graph-functions","title":"Functions","text":""},{"location":"dagpipe/#dagpipe.Graph.__repr__","title":"__repr__","text":"
Unique identifier of the node (snake_case dotted format).
namestr
Human-readable display name.
Notes
Responsibilities:
- Represents a deterministic unit of execution in the pipeline graph.\n- Consumes one `State` and produces zero, one, or many derived states.\n- Defines execution logic and enables branching, filtering, and transformation.\n
Guarantees:
- Nodes must never mutate the input `State`.\n- Instances are singletons per subclass and reused across executions.\n
Stateless subclasses (no parameterized __init__) share one singleton instance per class \u2014 matching the original dagpipe behaviour underpinning set_registry-style configuration. Subclasses that declare an __init__ requiring instance-state arguments get a fresh instance per construction so pipeline builders can inject per-run dependencies.
- Subclasses implement specific resolution behavior.\n- Must not mutate input state.\n- Should use `fork()` to create child states.\n- May yield zero states to terminate a branch.\n
Immutable hierarchical container with dot-path access.
Attributes:
Name Type Description _dataMapping[str, Any]
Immutable hierarchical data structure.
Notes
Responsibilities:
- Stores execution data used by `State`.\n- Supports efficient atomic updates without modifying existing instances.\n- `Payload` instances are fully thread-safe due to immutability.\n
Executable pipeline created from YAML configuration.
Attributes:
Name Type Description engineEngine
Execution engine responsible for running the pipeline.
state_clsType[State]
Dynamically created State subclass with configured schema.
initial_payloadPayload
Default payload used when execution begins.
Notes
Responsibilities:
- Encapsulates engine, state type, and initial payload.\n- Provides a simplified interface for executing configured pipelines.\n- Safe for concurrent execution if underlying nodes are thread-safe.\n
Lightweight progress payload emitted by engine step hooks.
Mirrors the imperative ProgressMessage used by the legacy orchestrator so callers can surface counts/lines/errors without coupling the engine to pydantic.
Immutable execution state propagated through dagpipe pipeline.
Attributes:
Name Type Description payloadPayload
Execution data container.
schemaClassVar[Schema]
Payload validation schema.
confidencefloat
Execution confidence score.
parentOptional[State]
Parent state reference.
depthint
Execution depth.
historyTuple[str, ...]
Ordered node execution lineage.
metadataDict[str, Any]
Execution metadata.
Notes
Responsibilities:
- Represents a complete execution snapshot at a specific point in\n pipeline traversal.\n- Fundamental unit of execution in `dagpipe`.\n- Fully thread-safe due to immutability.\n
Name Type Description Default payload_updateMapping[str, Any] | None
Dot-path updates applied to the payload.
Noneconfidence_deltafloat
Adjustment applied to current confidence.
0.0node_idstr | None
Identifier of the node creating this state.
Nonemetadata_updateMapping[str, Any] | None
Updates merged into state metadata.
None
Returns:
Name Type Description StateState
A new immutable State instance.
Notes
Guarantees:
- This is the only supported mechanism for modifying execution data.\n- Validates payload updates, preserves lineage, increments depth,\n and appends to history.\n
Execution engine responsible for running pipelines and graphs.
The Engine executes Node objects and propagates immutable State instances through either a linear sequence or a directed acyclic graph (Graph). It orchestrates execution order, branching, and state propagation.
Execution engine responsible for running pipeline logic.
Notes
Responsibilities:
- Accepts either a linear sequence of `Node` objects or a `Graph`\n defining execution topology.\n- Propagates immutable `State` objects through `Node` objects and\n collects terminal states.\n- Supports synchronous (`run`) and asynchronous (`run_async`)\n execution, dispatching per-node.\n- Supports step-wise / resumable execution and progress hooks.\n
Guarantees:
- Never mutates `State`, `Node`, or `Graph` instances.\n- `State` objects are never modified in place; each branch produces\n independent instances.\n- Execution order is deterministic and follows graph or pipeline topology.\n- Thread-safe for concurrent execution.\n
Execute the pipeline step-by-step, yielding one StepResult per step.
Parameters:
Name Type Description Default rootState
Initial execution state.
required resume_fromint | None
Skip steps at index < resume_from (for resume-after-partial). Steps are 0-indexed.
Noneon_stepStepHook | None
Callback (step, status, message) invoked per step; falls back to the engine-level hook when unset.
None
Yields:
Name Type Description StepResultStepResult
One per executed node/step, carrying the produced states.
Notes
This is a synchronous, generator-based checkpoint interface compatible with the imperative resume-by-step behaviour of the legacy orchestrator. Use run_steps_async for async nodes.
Lightweight progress payload emitted by engine step hooks.
Mirrors the imperative ProgressMessage used by the legacy orchestrator so callers can surface counts/lines/errors without coupling the engine to pydantic.
Directed Acyclic Graph defining execution topology of Node objects.
Notes
Responsibilities:
- Stores node connectivity and validates that the topology remains acyclic.\n- Structure determines how `State` flows between nodes during execution.\n
Guarantees:
- Topology is acyclic. Node relationships remain consistent.\n- Thread-safe for concurrent reads after construction.\n
Create an empty Graph.
"},{"location":"dagpipe/graph/#dagpipe.graph.Graph--initializes-node-registry-and-edge-mappings","title":"Initializes node registry and edge mappings.","text":""},{"location":"dagpipe/graph/#dagpipe.graph.Graph-functions","title":"Functions","text":""},{"location":"dagpipe/graph/#dagpipe.graph.Graph.__repr__","title":"__repr__","text":"
Base class for nodes whose execution is asynchronous.
Subclasses implement resolve_async (an async generator yielding derived State objects). The engine dispatches to resolve_async when running an async traversal (see Engine.run_async).
Sync-only engines (and the base Node.run) treat an AsyncNode as a no-op consumer: calling run on an AsyncNode returns no states, signalling that an async engine is required.
Unique identifier of the node (snake_case dotted format).
namestr
Human-readable display name.
Notes
Responsibilities:
- Represents a deterministic unit of execution in the pipeline graph.\n- Consumes one `State` and produces zero, one, or many derived states.\n- Defines execution logic and enables branching, filtering, and transformation.\n
Guarantees:
- Nodes must never mutate the input `State`.\n- Instances are singletons per subclass and reused across executions.\n
Stateless subclasses (no parameterized __init__) share one singleton instance per class \u2014 matching the original dagpipe behaviour underpinning set_registry-style configuration. Subclasses that declare an __init__ requiring instance-state arguments get a fresh instance per construction so pipeline builders can inject per-run dependencies.
- Subclasses implement specific resolution behavior.\n- Must not mutate input state.\n- Should use `fork()` to create child states.\n- May yield zero states to terminate a branch.\n
The State represents a single point in pipeline execution. It contains arbitrary data and metadata and is designed to be immutable. Instead of modifying an existing state, nodes create new child states via fork().
Immutable hierarchical container with dot-path access.
Attributes:
Name Type Description _dataMapping[str, Any]
Immutable hierarchical data structure.
Notes
Responsibilities:
- Stores execution data used by `State`.\n- Supports efficient atomic updates without modifying existing instances.\n- `Payload` instances are fully thread-safe due to immutability.\n
Immutable execution state propagated through dagpipe pipeline.
Attributes:
Name Type Description payloadPayload
Execution data container.
schemaClassVar[Schema]
Payload validation schema.
confidencefloat
Execution confidence score.
parentOptional[State]
Parent state reference.
depthint
Execution depth.
historyTuple[str, ...]
Ordered node execution lineage.
metadataDict[str, Any]
Execution metadata.
Notes
Responsibilities:
- Represents a complete execution snapshot at a specific point in\n pipeline traversal.\n- Fundamental unit of execution in `dagpipe`.\n- Fully thread-safe due to immutability.\n
Name Type Description Default payload_updateMapping[str, Any] | None
Dot-path updates applied to the payload.
Noneconfidence_deltafloat
Adjustment applied to current confidence.
0.0node_idstr | None
Identifier of the node creating this state.
Nonemetadata_updateMapping[str, Any] | None
Updates merged into state metadata.
None
Returns:
Name Type Description StateState
A new immutable State instance.
Notes
Guarantees:
- This is the only supported mechanism for modifying execution data.\n- Validates payload updates, preserves lineage, increments depth,\n and appends to history.\n
Creates fully configured pipeline objects from declarative YAML definitions, including Schema, State subclasses, Node instances, Graph topology, and initial payloads.
Executable pipeline created from YAML configuration.
Attributes:
Name Type Description engineEngine
Execution engine responsible for running the pipeline.
state_clsType[State]
Dynamically created State subclass with configured schema.
initial_payloadPayload
Default payload used when execution begins.
Notes
Responsibilities:
- Encapsulates engine, state type, and initial payload.\n- Provides a simplified interface for executing configured pipelines.\n- Safe for concurrent execution if underlying nodes are thread-safe.\n
Note: The extraction pipeline discussed below is a sample/example illustrating the hexa composition pattern.
-
-
This document is the concrete counterpart to philosophy.md: where
-the files are, what the tree looks like for both banks, and how a future container
-instantiates it. Self-sufficient for a fresh agent.
-
File map (samples/extraction_pipeline/)
-
-
-
-
File
-
Layer
-
Role
-
-
-
-
-
extraction_pipeline_abc.py
-
ABC contracts
-
All port ABCs + ExtractionPipeline root. Declares attributes, the ExtractionSource dataclass, port slots, @abstractmethods.
-
-
-
extraction_pipeline_impl.py
-
Generic Impl*
-
Concrete default implementations of every port; ImplExtractionPipeline root; clean_num helper.
-
-
-
extraction_pipeline_ambiguity.py
-
Shared logic
-
Ported ResolutionContext, AmbiguityHandler, AmountBalanceNotFound, validate_signs — used by ImplAmountBalancePort.
AxisTxnDictsPort(ImplTxnDictsPort)
-└─parser:AxisTransactionParserPort(ImplTransactionParserPort)
-├─number:AxisNumberPort# NUMBER_RE, all clean numbers
-├─desc:AxisDescPort# STARTTERS = [upi/, imps/, neft-, neft/, ach/, ach-, 2a/]
-└─amount_balance:AxisAmountBalancePort(ImplAmountBalancePort)
-└─number:AxisNumberPort# re-pin back to axis number port
-
-
Notes:
-- date is not redeclared on AxisTransactionParserPort — it inherits
- ImplDatePort (correct already, adds no behavior).
-- amount_balance.ambiguity is not redeclared — inherits ImplAmbiguityPort.
Only redeclare a slot you are actually changing from Impl*. If a slot's
-behavior is already correct from the parent Impl*, omit it. This keeps the
-diff between a bank and Impl* minimal and unambiguous.
-
-
date: ImplDatePort was previously redeclared redundantly and has been removed —
-it added mental load on the extender for zero behavior change. (Fidelity audit:
-it is still concrete ImplDatePort in ImplTransactionParserPort, so behavior is
-unchanged.)
-
-
Container (build(cls) in hexa/container.py)
-
The annotations are metadata; a small reflection builder turns a root class into
-a wired instance:
Walk cls.__annotations__, merged across the MRO so inherited slot annotations
- are visible (a bank subclass only adds/overrides a slot; the rest come from
- Impl*).
-
For each port slot whose annotated type is a concrete Impl*/bank class
- (not an ABC, not a builtin), instantiate it.
-
Recurse into that child — repeat until leaves (classes that declare no further
- port slots).
-
setattr(parent, slot, child) to wire each slot onto the parent instance.
-
Apply config defaults from the class-level annotated attributes.
-
-
Selection is fully static: the annotated type is the chosen implementation.
-The container holds no per-bank branch logic — Axis vs Icici is decided purely by
-which root class you call build on.
-
Control flow at runtime (target)
-
A leaf-stage port exposes one entry method; stages chain data through the shared
-run(source) signature on the root:
The stage ports map naturally onto dagpipe Nodes (ingest, raw_lines,
-txn_blocks, txn_dicts, raw_expense), each node's resolve delegating to its
-wired port subtree, connected as a dagpipe Graph and run via Engine. That merge
-is a separate, future step.
-
Utilities (optional, hexa-agnostic)
-
The hexa utilities are implemented in hexa/ (the package root) and they are agnostic of the extraction pipeline sample. It ships a set of optional code-gen / validation utilities so users can choose their own workflow — they are never required. A user is free to:
-
-
hand-write an ABC file and never touch YAML, or
-
start from YAML and generate the ABC file, or
-
start from an ABC file and emit YAML for it, or
-
use the ABC/YAML only as a spec (neither generated from the other).
-
-
The utilities only ever translate between two interchangeable representations of
-the same contract — the .yaml spec and the _abc.py module. They never
-generate or manage the Impl*/bank files (those carry behavior and slot choice,
-which is a human decision).
-
Shared model
-
All utilities serialize through one neutral model, a PortNode tree. It is NOT
-tied to the extraction sample — any port tree fits.
@dataclass
-classPortNode:
-name:str# slot name, e.g. "parser", "amount_balance"
-kind:str# "leaf" | "port" | "parent"
-port_cls:str# ABC class name, e.g. "TransactionParserPort"
-attributes:dict[str,ConfigValue]# annotated attributes: name -> type expr,
-# or a nested struct dict for grouped values
-children:list["PortNode"]
-methods:dict[str,MethodSpec]# abstract methods
-
-
1. parse_yaml(path) -> PortNode
-
Read a YAML spec into the PortNode tree (YAML → model).
-
-
Parses completely (structural validation only, no type-checking).
-
Supports the sample grammar (see extraction_pipeline.yaml): a root class whose
- inline attributes (scalars and nested structs), port slots, and reserved
- methods: recurse into child ports.
-
Raises ValueError with file/line on structural errors.
-
-
2. generate_abc(path_yaml, out_path=None) -> str
-
Generate _abc.py source from a YAML spec (YAML → ABC).
-
-
parse_yaml(...) → write one class <PortCls>(ABC) per PortNode with
- attribute annotations (nested structs become synthesized @dataclasses), port-slot
- annotations, and @abstractmethod stubs, matching the style of the hand-written
- extraction_pipeline_abc.py.
-
Deterministic (YAML-order) output so it is idempotent / diff-friendly.
-
out_path optional; always returns the generated source.
-
-
3. parse_abc(module_or_path) -> PortNode
-
Reflect over an ABC module into the PortNode tree (ABC → model).
-
-
Reads __annotations__ / @abstractmethod from generic ABC classes (no
- sample assumptions) to reconstruct the same PortNode tree.
Generate a .yaml spec from an ABC module (ABC → YAML).
-
-
parse_abc(...) → write the reverse of generate_abc.
-
Lets a user author the ABC file first and emit YAML for it.
-
-
5. check_matches(path_a, path_b) -> bool
-
Verify two representations agree — YAML vs ABC, either direction.
-
-
Builds both PortNode models (parse_yaml + parse_abc) and compares: classes
- exist and are ABCs, attributes match, port slots match, @abstractmethod
- signatures match (missing keyword-only params treated conservatively).
-
Returns True only if the whole tree matches; otherwise False (or a diff).
Hexa is a pattern and utility library for structuring complex hierarchical pipelines. It rests on one core idea: a concrete class's annotated slot type is the dependency decision. Variation is expressed entirely as class annotations \u2014 never as imperative wiring, never as __init__ parameters, never as a composition root that assembles objects by hand.
All utilities serialize through a single shared model (PortNode):
parse_yaml / generate_yaml \u2014 between YAML specs and the model
parse_abc / generate_abc \u2014 between ABC Python modules and the model
check_matches \u2014 verify two representations describe the same tree
build \u2014 recursively instantiate a wired pipeline from annotations
Each node is a leaf (no children), a port (has nested port slots), or the parent root of the tree. Every node carries a concrete typed shape derived from an ABC class name.
Attributes:
Name Type Description namestr
Slot name, e.g. \"parser\", \"amount_balance\".
port_clsstr
ABC class name that gives this port its shape, e.g. \"TransactionParserPort\". Every port has a concrete typed shape.
kindstr
One of \"leaf\", \"port\", or \"parent\". - \"leaf\": no children (e.g. NumberPort) - \"port\": has children (e.g. TransactionParserPort) - \"parent\": the root of the tree (e.g. ExtractionPipeline)
attributesdict[str, ConfigValue]
Annotated attributes: name -> type_expr for scalars, or a nested dict of sub-attributes (a struct) for grouped values.
Recursively instantiate the port tree from annotations.
Parameters:
Name Type Description Default clstype[T]
The root class to build (e.g. AxisExtractionPipeline).
required instancesdict[str, Any] | None
{slot_name: object} injected verbatim into every node whose annotations contain that name (runtime values \u2014 repos, handlers, clients \u2014 that must not be re-constructed). Applied to the whole tree by name.
Noneconfigdict[str, Any] | None
{field_name: value} propagated by name to every node's annotated config fields (non-port attributes), replacing the default None.
None
Returns:
Name Type Description TT
A fully wired instance of cls and all its nested ports.
check_matches parses two representations \u2014 one YAML spec and/or ABC Python module each \u2014 and reports whether they describe the same port tree.
Recursively instantiate the port tree from annotations.
Parameters:
Name Type Description Default clstype[T]
The root class to build (e.g. AxisExtractionPipeline).
required instancesdict[str, Any] | None
{slot_name: object} injected verbatim into every node whose annotations contain that name (runtime values \u2014 repos, handlers, clients \u2014 that must not be re-constructed). Applied to the whole tree by name.
Noneconfigdict[str, Any] | None
{field_name: value} propagated by name to every node's annotated config fields (non-port attributes), replacing the default None.
None
Returns:
Name Type Description TT
A fully wired instance of cls and all its nested ports.
Note: The extraction pipeline discussed below is a sample/example illustrating the hexa composition pattern.
This document is the concrete counterpart to philosophy.md: where the files are, what the tree looks like for both banks, and how a future container instantiates it. Self-sufficient for a fresh agent.
"},{"location":"design/#file-map-samplesextraction_pipeline","title":"File map (samples/extraction_pipeline/)","text":"File Layer Role extraction_pipeline_abc.py ABC contracts All port ABCs + ExtractionPipeline root. Declares attributes, the ExtractionSource dataclass, port slots, @abstractmethods. extraction_pipeline_impl.py Generic Impl* Concrete default implementations of every port; ImplExtractionPipeline root; clean_num helper. extraction_pipeline_ambiguity.py Shared logic Ported ResolutionContext, AmbiguityHandler, AmountBalanceNotFound, validate_signs \u2014 used by ImplAmountBalancePort. extraction_pipeline_banks.py Composition root AxisExtractionPipeline(ImplExtractionPipeline) + IciciExtractionPipeline (each re-pins txn_dicts). banks/axis/pdf.py Bank specialization AxisNumberPort, AxisDescPort, AxisAmountBalancePort, AxisTransactionParserPort, AxisTxnDictsPort. banks/icici/pdf.py Bank specialization Icici* mirror of the above + Icici-only adjust_balance/missing_number_candidates. extraction_pipeline.yaml Contract spec Declarative spec of the port tree; interchangeable with _abc.py via the optional utilities."},{"location":"design/#the-port-tree-abc-contract","title":"The port tree (ABC contract)","text":"
class AxisExtractionPipeline(ImplExtractionPipeline):\n txn_dicts: AxisTxnDictsPort\n
AxisTxnDictsPort(ImplTxnDictsPort)\n \u2514\u2500 parser: AxisTransactionParserPort(ImplTransactionParserPort)\n \u251c\u2500 number: AxisNumberPort # NUMBER_RE, all clean numbers\n \u251c\u2500 desc: AxisDescPort # STARTTERS = [upi/, imps/, neft-, neft/, ach/, ach-, 2a/]\n \u2514\u2500 amount_balance: AxisAmountBalancePort(ImplAmountBalancePort)\n \u2514\u2500 number: AxisNumberPort # re-pin back to axis number port\n
Notes: - date is not redeclared on AxisTransactionParserPort \u2014 it inherits ImplDatePort (correct already, adds no behavior). - amount_balance.ambiguity is not redeclared \u2014 inherits ImplAmbiguityPort.
"},{"location":"design/#redundancy-rule-applied-in-both-banks","title":"Redundancy rule (applied in both banks)","text":"
Only redeclare a slot you are actually changing from Impl*. If a slot's behavior is already correct from the parent Impl*, omit it. This keeps the diff between a bank and Impl* minimal and unambiguous.
date: ImplDatePort was previously redeclared redundantly and has been removed \u2014 it added mental load on the extender for zero behavior change. (Fidelity audit: it is still concrete ImplDatePort in ImplTransactionParserPort, so behavior is unchanged.)
"},{"location":"design/#container-buildcls-in-hexacontainerpy","title":"Container (build(cls) in hexa/container.py)","text":"
The annotations are metadata; a small reflection builder turns a root class into a wired instance:
Walk cls.__annotations__, merged across the MRO so inherited slot annotations are visible (a bank subclass only adds/overrides a slot; the rest come from Impl*).
For each port slot whose annotated type is a concrete Impl*/bank class (not an ABC, not a builtin), instantiate it.
Recurse into that child \u2014 repeat until leaves (classes that declare no further port slots).
setattr(parent, slot, child) to wire each slot onto the parent instance.
Apply config defaults from the class-level annotated attributes.
Selection is fully static: the annotated type is the chosen implementation. The container holds no per-bank branch logic \u2014 Axis vs Icici is decided purely by which root class you call build on.
"},{"location":"design/#control-flow-at-runtime-target","title":"Control flow at runtime (target)","text":"
A leaf-stage port exposes one entry method; stages chain data through the shared run(source) signature on the root:
The stage ports map naturally onto dagpipe Nodes (ingest, raw_lines, txn_blocks, txn_dicts, raw_expense), each node's resolve delegating to its wired port subtree, connected as a dagpipe Graph and run via Engine. That merge is a separate, future step.
The hexa utilities are implemented in hexa/ (the package root) and they are agnostic of the extraction pipeline sample. It ships a set of optional code-gen / validation utilities so users can choose their own workflow \u2014 they are never required. A user is free to:
hand-write an ABC file and never touch YAML, or
start from YAML and generate the ABC file, or
start from an ABC file and emit YAML for it, or
use the ABC/YAML only as a spec (neither generated from the other).
The utilities only ever translate between two interchangeable representations of the same contract \u2014 the .yaml spec and the _abc.py module. They never generate or manage the Impl*/bank files (those carry behavior and slot choice, which is a human decision).
Read a YAML spec into the PortNode tree (YAML \u2192 model).
Parses completely (structural validation only, no type-checking).
Supports the sample grammar (see extraction_pipeline.yaml): a root class whose inline attributes (scalars and nested structs), port slots, and reserved methods: recurse into child ports.
Raises ValueError with file/line on structural errors.
Generate _abc.py source from a YAML spec (YAML \u2192 ABC).
parse_yaml(...) \u2192 write one class <PortCls>(ABC) per PortNode with attribute annotations (nested structs become synthesized @dataclasses), port-slot annotations, and @abstractmethod stubs, matching the style of the hand-written extraction_pipeline_abc.py.
Deterministic (YAML-order) output so it is idempotent / diff-friendly.
out_path optional; always returns the generated source.
Verify two representations agree \u2014 YAML vs ABC, either direction.
Builds both PortNode models (parse_yaml + parse_abc) and compares: classes exist and are ABCs, attributes match, port slots match, @abstractmethod signatures match (missing keyword-only params treated conservatively).
Returns True only if the whole tree matches; otherwise False (or a diff).
Each node is a leaf (no children), a port (has nested port slots), or the parent root of the tree. Every node carries a concrete typed shape derived from an ABC class name.
Attributes:
Name Type Description namestr
Slot name, e.g. \"parser\", \"amount_balance\".
port_clsstr
ABC class name that gives this port its shape, e.g. \"TransactionParserPort\". Every port has a concrete typed shape.
kindstr
One of \"leaf\", \"port\", or \"parent\". - \"leaf\": no children (e.g. NumberPort) - \"port\": has children (e.g. TransactionParserPort) - \"parent\": the root of the tree (e.g. ExtractionPipeline)
attributesdict[str, ConfigValue]
Annotated attributes: name -> type_expr for scalars, or a nested dict of sub-attributes (a struct) for grouped values.
Parse an ABC Python module into a hexa PortNode tree.
ABC classes become port nodes. A nested-struct attribute is expressed as a @dataclass annotation on an ABC port (e.g. source: ExtractionSource where ExtractionSource is a @dataclass); the dataclass's fields are expanded into a nested attributes entry.
Parse an ABC Python module into a :class:PortNode tree.
ABC classes become port nodes. A nested-struct attribute is expressed as a @dataclass annotation on an ABC port (e.g. source: ExtractionSource where ExtractionSource is a @dataclass); the dataclass's fields are expanded into a nested attributes entry.
Parameters:
Name Type Description Default module_or_pathstr | Path | Any
A file path to a .py file, a module import string, or a loaded module.
Parse a hexa YAML spec into a :class:PortNode tree.
The YAML grammar uses a nested mapping format built from three member kinds:
attribute: a name: type scalar (e.g. source_type: str)
nested struct: a name: mapping of sub-attributes (e.g. source:)
port: a name: mapping whose single key is a class whose value is a body, or a class key with a ... body. A port carries a class shape.
Disambiguation of a mapping value
A mapping with exactly one key whose value is a non-scalar body (a mapping, ..., or a list) is a port. Any other mapping (multiple keys, or all scalar-typed values) is a nested struct of attributes.
This document explains the why behind the type-declared dependency pattern used for the extraction pipeline sample \u2014 specifically the Axis/Icici bank variation. It is, together with design.md, self-sufficient: a fresh agent should be able to pick up the codebase from here and reason about (and extend) the pattern without needing the original session context.
A concrete class's annotated slot type is the dependency decision.
Bank variation is expressed entirely as class annotations \u2014 never as imperative wiring, never as __init__ parameters, never as a composition-root that assembles objects by hand. The tree of ports a pipeline needs is declared once, and a bank differs from the generic pipeline only by which concrete types are pinned onto which slots.
"},{"location":"philosophy/#the-three-layers","title":"The three layers","text":"
The pattern is built from three fixed layers. Understanding which layer something belongs to is the whole mental model.
Authoritative declaration of need (\"WHAT a stage requires\").
Each port is an ABC.
It declares attributes as class-level annotated defaults (e.g. TransactionParserPort.min_numbers: int = 3), optionally grouped as a nested struct via a @dataclass (e.g. source: ExtractionSource).
It declares port slots as annotated attributes whose type is another ABC (e.g. TransactionParserPort.number: NumberPort, amount_balance: AmountBalancePort).
It declares @abstractmethod bodies \u2014 the method signatures each concrete implementation must provide.
The ABC never runs, never holds an instance, and never says which implementation to use. It only says what the shape is.
The common default behavior (\"Imple \u2014 the shared adaptor\").
class ImplTransactionParserPort(TransactionParserPort) subclasses the ABC, implements every @abstractmethod, and re-pins its port slots to concrete Impl* types: number: ImplNumberPort, desc: ImplDescPort, amount_balance: ImplAmountBalancePort.
ImplAmountBalancePort in turn pins number: ImplNumberPort and ambiguity: ImplAmbiguityPort.
The root, ImplExtractionPipeline, pins its five stage slots to Impl* ports: ingest, raw_lines, txn_blocks, txn_dicts, raw_expense.
Impl* is the default that shared, bank-agnostic behavior lives in. Any slot a bank does not override falls back to Impl* at runtime.
"},{"location":"philosophy/#layer-3-bank-specializations-samplesextraction_pipelinebanksaxisicicipdfpy-samplesextraction_pipelineextraction_pipeline_bankspy","title":"Layer 3 \u2014 bank specializations (samples/extraction_pipeline/banks/{axis,icici}/pdf.py, samples/extraction_pipeline/extraction_pipeline_banks.py)","text":"
The variation (\"what differs per bank\").
Bank classes subclass the concrete Impl* classes, never the ABC \u2014 they extend, they never reinvent the contract.
Each Axis*/Icici* class overrides only the members that differ from the generic Impl*; everything shared is inherited.
A bank root is a thin annotated subclass of ImplExtractionPipeline that re-pins only the slots that differ:
class AxisExtractionPipeline(ImplExtractionPipeline):\n txn_dicts: AxisTxnDictsPort\n
Bank-specific wiring is hierarchical and cascading. Each level narrows exactly one slot, and its narrowed type drags the next level's narrowing along with it:
AxisExtractionPipeline\n \u2514\u2500 txn_dicts: AxisTxnDictsPort (only override on the root)\n \u2514\u2500 parser: AxisTransactionParserPort\n \u251c\u2500 number: AxisNumberPort (get_numbers \u2192 all clean nums)\n \u251c\u2500 desc: AxisDescPort (STARTTERS)\n \u2514\u2500 amount_balance: AxisAmountBalancePort\n \u2514\u2500 number: AxisNumberPort (re-pin: same bank's number port)\n
Siblings that are already correct in Impl* are inherited, not redeclared. E.g. the parser's date slot comes from ImplTransactionParserPort; the amount-balance port's ambiguity comes from ImplAmountBalancePort. Re-declaring them adds mental load on the extender with zero behavior change.
Rule \u2014 only redeclare the slot you are actually changing. If a slot's behavior is already correct from Impl*, leave it out.
"},{"location":"philosophy/#the-two-fidelity-guarantees-this-pattern-preserves","title":"The two fidelity guarantees this pattern preserves","text":"
Shared behavior stays shared. Slots like ingest, raw_lines, txn_blocks, raw_expense, parser date, and ambiguity are bank-agnostic. Leaving them at the Impl* default is deliberate, not an omission.
Banks never reinvent. Axis*/Icici* only extend Impl*. The ABC contract in samples/extraction_pipeline/extraction_pipeline_abc.py is defined once and shared by every bank.
"},{"location":"philosophy/#what-this-pattern-is-not","title":"What this pattern is NOT","text":"
No imperative assembly. No build_axis_pipeline() wiring objects by hand.
No constructor injection. Impl* classes have no __init__; wiring is done by assigning concrete port instances onto annotated slots.
YAML and ABCs are interchangeable, and both optional. The .yaml spec and the _abc.py module are two representations of the same contract. hexa is agnostic of the extraction pipeline sample and never imposes a workflow: a user can hand-write the ABC file, start from YAML (ABC generated), start from an ABC file (YAML emitted), or use both as a spec. The optional utilities in design.md \u00a7 Utilities translate between them and can verify they agree \u2014 but the bank Impl*/specialization files are always hand-written (their slots are a manual choice) and never generated.
The annotations are the decision; hexa.build(cls) (see design.md \u00a7 Container) turns a root class into a wired instance by reading __annotations__ + MRO, recursively instantiating each pinned concrete type, and setattring the child onto the parent slot. This document is scoped to the notation \u2014 the rest of the tree can be expressed purely as class annotations and filled at runtime by the container.
"}]}
\ No newline at end of file
+{"config":{"lang":["en"],"separator":"[\\s\\-]+","pipeline":["stopWordFilter"]},"docs":[{"location":"","title":"hexa","text":""},{"location":"#hexa","title":"hexa","text":""},{"location":"#hexa--summary","title":"Summary","text":"
Type-declared dependency injection for Python.
Hexa is a pattern and utility library for structuring complex hierarchical pipelines. It rests on one core idea: a concrete class's annotated slot type is the dependency decision. Variation is expressed entirely as class annotations \u2014 never as imperative wiring, never as __init__ parameters, never as a composition root that assembles objects by hand.
All utilities serialize through a single shared model (PortNode):
parse_yaml / generate_yaml \u2014 between YAML specs and the model
parse_abc / generate_abc \u2014 between ABC Python modules and the model
check_matches \u2014 verify two representations describe the same tree
build \u2014 recursively instantiate a wired pipeline from annotations
Each node is a leaf (no children), a port (has nested port slots), or the parent root of the tree. Every node carries a concrete typed shape derived from an ABC class name.
Attributes:
Name Type Description namestr
Slot name, e.g. \"parser\", \"amount_balance\".
port_clsstr
ABC class name that gives this port its shape, e.g. \"TransactionParserPort\". Every port has a concrete typed shape.
kindstr
One of \"leaf\", \"port\", or \"parent\". - \"leaf\": no children (e.g. NumberPort) - \"port\": has children (e.g. TransactionParserPort) - \"parent\": the root of the tree (e.g. ExtractionPipeline)
attributesdict[str, ConfigValue]
Annotated attributes: name -> type_expr for scalars, or a nested dict of sub-attributes (a struct) for grouped values.
Recursively instantiate the port tree from annotations.
Parameters:
Name Type Description Default clstype[T]
The root class to build (e.g. AxisExtractionPipeline).
required instancesdict[str, Any] | None
{slot_name: object} injected verbatim into every node whose annotations contain that name (runtime values \u2014 repos, handlers, clients \u2014 that must not be re-constructed). Applied to the whole tree by name.
Noneconfigdict[str, Any] | None
{field_name: value} propagated by name to every node's annotated config fields (non-port attributes), replacing the default None.
None
Returns:
Name Type Description TT
A fully wired instance of cls and all its nested ports.
check_matches parses two representations \u2014 one YAML spec and/or ABC Python module each \u2014 and reports whether they describe the same port tree.
Recursively instantiate the port tree from annotations.
Parameters:
Name Type Description Default clstype[T]
The root class to build (e.g. AxisExtractionPipeline).
required instancesdict[str, Any] | None
{slot_name: object} injected verbatim into every node whose annotations contain that name (runtime values \u2014 repos, handlers, clients \u2014 that must not be re-constructed). Applied to the whole tree by name.
Noneconfigdict[str, Any] | None
{field_name: value} propagated by name to every node's annotated config fields (non-port attributes), replacing the default None.
None
Returns:
Name Type Description TT
A fully wired instance of cls and all its nested ports.
Each node is a leaf (no children), a port (has nested port slots), or the parent root of the tree. Every node carries a concrete typed shape derived from an ABC class name.
Attributes:
Name Type Description namestr
Slot name, e.g. \"parser\", \"amount_balance\".
port_clsstr
ABC class name that gives this port its shape, e.g. \"TransactionParserPort\". Every port has a concrete typed shape.
kindstr
One of \"leaf\", \"port\", or \"parent\". - \"leaf\": no children (e.g. NumberPort) - \"port\": has children (e.g. TransactionParserPort) - \"parent\": the root of the tree (e.g. ExtractionPipeline)
attributesdict[str, ConfigValue]
Annotated attributes: name -> type_expr for scalars, or a nested dict of sub-attributes (a struct) for grouped values.
Parse an ABC Python module into a hexa PortNode tree.
ABC classes become port nodes. A nested-struct attribute is expressed as a @dataclass annotation on an ABC port (e.g. source: ExtractionSource where ExtractionSource is a @dataclass); the dataclass's fields are expanded into a nested attributes entry.
Parse an ABC Python module into a :class:PortNode tree.
ABC classes become port nodes. A nested-struct attribute is expressed as a @dataclass annotation on an ABC port (e.g. source: ExtractionSource where ExtractionSource is a @dataclass); the dataclass's fields are expanded into a nested attributes entry.
Parameters:
Name Type Description Default module_or_pathstr | Path | Any
A file path to a .py file, a module import string, or a loaded module.
Parse a hexa YAML spec into a :class:PortNode tree.
The YAML grammar uses a nested mapping format built from three member kinds:
attribute: a name: type scalar (e.g. source_type: str)
nested struct: a name: mapping of sub-attributes (e.g. source:)
port: a name: mapping whose single key is a class whose value is a body, or a class key with a ... body. A port carries a class shape.
Disambiguation of a mapping value
A mapping with exactly one key whose value is a non-scalar body (a mapping, ..., or a list) is a port. Any other mapping (multiple keys, or all scalar-typed values) is a nested struct of attributes.
hexa lets you describe a hierarchical pipeline as a tree of typed contracts, then get a fully wired instance back from plain class annotations. This page builds the mental model: the three layers, the two interchangeable representations, and the container that reads annotations at runtime.
Representation — the same contract can be written as a YAML spec or as a Python ABC module. hexa's converters translate between them through one neutral PortNode tree. This layer is optional: you may hand-write only the ABC file, only the YAML, or both.
+
Runtime — build(cls) reads the merged annotations of a root class. Every port slot annotated with a concrete type is the decision; the container instantiates the tree recursively and setattrs each child onto its parent.
A validated reference to the public API of hexa (package root exports from hexa/__init__.py). Signatures match the code exactly. For runnable recipes see the Use Cases. Exact parameter contracts live in the API reference under docs/lib/ — this page is usage-level.
Specification of a single abstract method: name, positional args: list[str], and keyword-only kwargs: dict[str, str] (name → type expression). Supports equality (==/!=).
str | NestedConfig, where NestedConfig = dict[str, ConfigValue]. An attribute is either a scalar type expression (e.g. "str") or a nested structure of sub-attributes. Nesting is recursive.
Parses a hexa YAML spec into a PortNode tree. The grammar uses three member kinds:
+
+
attribute — name: type scalar (e.g. source_type: str)
+
nested struct — name: mapping of sub-attributes (e.g. source:)
+
port — name: mapping whose single key is a class whose value is a body (e.g. ingest: IngestPort: ...)
+
+
Disambiguation rule: a mapping with exactly one key whose value is a non-scalar body is a port; any other mapping is a nested struct of attributes. Raises ValueError on structural errors (see Error Handling).
Generates _abc.py source from a PortNode tree. Emits class X(ABC) per node, synthesized @dataclasses for nested structs, port-slot annotations, and @abstractmethod stubs. Deterministic (walk order) so output is idempotent. When out_path is given the source is also written there; the source string is always returned.
Reflects an ABC Python module into a PortNode tree. Accepts a .py file path, a module import string, or an already-loaded module. ABC classes become nodes; a @dataclass annotation on a port expands into a nested attributes entry. The root is auto-detected (the class not referenced as any other slot, preferring names containing Pipeline/Root/Parent).
check_matches(path_a, path_b) -> bool | list[str] parses two representations (one .yaml/.yml and one .py, either direction) and compares the trees with PortNode.diff. Returns True when they describe the same tree, otherwise a list of human-readable difference strings. Inequality checks cover port_cls, kind, attributes (including nested structs), methods (args/kwargs), and children.
Scenario: you design the contract as a YAML spec, generate ABC Python source from it, hand-write concrete implementations, and prove the spec and code agree.
generate_abc emits contracts only — it never generates Impl*/specialization files. Slot pinning and method bodies are your call.
+
The YAML grammar distinguishes a port (a mapping whose single key has a non-scalar body) from a nested struct (any other mapping). Keep exactly one class key per port slot.
+
check_matches compares classes, attributes, slots, and @abstractmethod signatures; missing keyword-only params are treated conservatively so minor refactors don't false-negative.
+
All generators are deterministic (walk order), so regenerating a spec from a module and diffing is stable.
Author the contract in Python (see samples/minimal/abc.py). A nested-struct attribute is expressed as a @dataclass annotation on an ABC port; the dataclass fields expand into a nested attributes entry when parsed:
Root detection: parse_abc picks the class that is not referenced as any other slot; on ties it prefers names containing Pipeline, Root, or Parent.
+
parse_abc accepts a .py path, an import string (e.g. samples.minimal.abc), or an already-loaded module, so it composes well in scripts.
+
A @dataclass referenced from a port becomes a nested struct; a dataclass field that references an ABC becomes a nested port slot keyed under the struct.
+
YAML generated this way is exactly the grammar parse_yaml understands — the two directions round-trip (node1.diff(node3) == [] in the repo's roundtrip tests).
Scenario: you have concrete classes with annotated slots and you want a fully wired instance — including runtime values (repos, clients) and config values (retries, labels, thresholds) — without writing any assembly code.
This mirrors tests/test_container.py exactly. Define leaf, mid-level, and root classes purely with annotations:
+
Python
fromhexaimportbuild
+
+
+class_Repo:
+def__init__(self):
+self.name="injected-repo"
+
+
+class_Leaf:
+field:int=0
+
+
+class_Mid:
+leaf:_Leaf
+threshold:float=0.0
+
+
+class_Root:
+repo:_Repo
+mid:_Mid
+retries:int=0
+label:str=""
+
+
+# Provide a runtime instance for a slot, and config values by name.
+repo=_Repo()
+root=build(
+_Root,
+instances={"repo":repo},
+config={"retries":3,"label":"fast","threshold":2.5},
+)
+
+assertroot.repoisrepo# injected verbatim, not reconstructed
+assertroot.retries==3# config applied
+assertroot.label=="fast"
+assertroot.mid.leafisnotNone# nested ports built recursively
+assertroot.mid.threshold==2.5# config propagates by name into nested nodes
+
+
Without any overrides, the same annotations still produce a working tree — every slot is instantiated from its annotated type and config fields keep their class defaults:
Optional / typing-generic config fields are handled the same way — a str | None annotated attribute takes its config value when provided and defaults to the class default otherwise:
+
Python
class_OptionalConfig:
+trust_fallback:str|None=None
+starters:list|None=None
+port:_Leaf
+
+
+built=build(_OptionalConfig,config={"trust_fallback":"amount"})
+assertbuilt.trust_fallback=="amount"
+assertbuilt.startersisNone# unrelated optional field keeps its default
+
instances= wins over reconstruction. If a name is in instances, its object is setattred verbatim at every node that annotates it. Use it for things that must not be re-created (repos, handlers, clients).
+
config= targets non-port annotated fields. Builtin, typing, and PEP 604-union (X | None) annotated fields are treated as config, applied by name and propagated to every node.
+
Abstract classes are never instantiated; they take a config value if one is provided (useful for defaults on CLI-style options).
+
Only concrete annotated types become wired ports. String forward references are skipped (the container assumes Impl files pin actual types).
+
Port instances are wired by setattr(parent, slot, child) — so methods can reference self.portX safely once the root is built.
+
If cls() raises TypeError, build retries by passing None for each declared __init__ parameter — convenient for classes with optional constructor args.
Note: The extraction pipeline discussed below is a sample/example illustrating the hexa composition pattern.
+
+
This document is the concrete counterpart to Philosophy: where the files are, what the tree looks like for both banks, and how a future container instantiates it. Self-sufficient for a fresh agent.
Only redeclare a slot you are actually changing from Impl*. If a slot's behavior is already correct from the parent Impl*, omit it. This keeps the diff between a bank and Impl* minimal and unambiguous.
+
+
date: ImplDatePort was previously redeclared redundantly and has been removed — it added mental load on the extender for zero behavior change. (Fidelity audit: it is still concrete ImplDatePort in ImplTransactionParserPort, so behavior is unchanged.)
Walk cls.__annotations__, merged across the MRO so inherited slot annotations are visible (a bank subclass only adds/overrides a slot; the rest come from Impl*).
+
For each port slot whose annotated type is a concrete Impl*/bank class (not an ABC, not a builtin), instantiate it.
+
Recurse into that child — repeat until leaves (classes that declare no further port slots).
+
setattr(parent, slot, child) to wire each slot onto the parent instance.
+
Apply config defaults from the class-level annotated attributes.
+
+
Selection is fully static: the annotated type is the chosen implementation. The container holds no per-bank branch logic — Axis vs Icici is decided purely by which root class you call build on.
The stage ports map naturally onto dagpipe Nodes (ingest, raw_lines, txn_blocks, txn_dicts, raw_expense), each node's resolve delegating to its wired port subtree, connected as a dagpipe Graph and run via Engine. That merge is a separate, future step.
The hexa utilities are implemented in hexa/ (the package root) and they are agnostic of the extraction pipeline sample. It ships a set of optional code-gen / validation utilities so users can choose their own workflow — they are never required. A user is free to:
+
+
hand-write an ABC file and never touch YAML, or
+
start from YAML and generate the ABC file, or
+
start from an ABC file and emit YAML for it, or
+
use the ABC/YAML only as a spec (neither generated from the other).
+
+
The utilities only ever translate between two interchangeable representations of the same contract — the .yaml spec and the _abc.py module. They never generate or manage the Impl*/bank files (those carry behavior and slot choice, which is a human decision).
Read a YAML spec into the PortNode tree (YAML → model).
+
+
Parses completely (structural validation only, no type-checking).
+
Supports the sample grammar (see extraction_pipeline.yaml): a root class whose inline attributes (scalars and nested structs), port slots, and reserved methods: recurse into child ports.
+
Raises ValueError with file/line on structural errors.
Generate _abc.py source from a YAML spec (YAML → ABC).
+
+
parse_yaml(...) → write one class <PortCls>(ABC) per PortNode with attribute annotations (nested structs become synthesized @dataclasses), port-slot annotations, and @abstractmethod stubs, matching the style of the hand-written extraction_pipeline_abc.py.
+
Deterministic (YAML-order) output so it is idempotent / diff-friendly.
+
out_path optional; always returns the generated source.
Verify two representations agree — YAML vs ABC, either direction.
+
+
Builds both PortNode models (parse_yaml + parse_abc) and compares: classes exist and are ABCs, attributes match, port slots match, @abstractmethod signatures match (missing keyword-only params treated conservatively).
+
Returns True only if the whole tree matches; otherwise False (or a diff).
This document explains the why behind the type-declared dependency pattern used
-for the extraction pipeline sample — specifically the Axis/Icici bank variation. It is,
-together with design.md, self-sufficient: a fresh agent should be able
-to pick up the codebase from here and reason about (and extend) the pattern without
-needing the original session context.
This document explains the why behind the type-declared dependency pattern used for the extraction pipeline sample — specifically the Axis/Icici bank variation. It is, together with Design, self-sufficient: a fresh agent should be able to pick up the codebase from here and reason about (and extend) the pattern without needing the original session context.
A concrete class's annotated slot type is the dependency decision.
-
Bank variation is expressed entirely as class annotations — never as imperative
-wiring, never as __init__ parameters, never as a composition-root that assembles
-objects by hand. The tree of ports a pipeline needs is declared once, and a bank
-differs from the generic pipeline only by which concrete types are pinned onto
-which slots.
-
The three layers
-
The pattern is built from three fixed layers. Understanding which layer something
-belongs to is the whole mental model.
Bank variation is expressed entirely as class annotations — never as imperative wiring, never as __init__ parameters, never as a composition-root that assembles objects by hand. The tree of ports a pipeline needs is declared once, and a bank differs from the generic pipeline only by which concrete types are pinned onto which slots.
Authoritative declaration of need ("WHAT a stage requires").
Each port is an ABC.
-
It declares attributes as class-level annotated defaults
- (e.g. TransactionParserPort.min_numbers: int = 3), optionally grouped as a
- nested struct via a @dataclass (e.g. source: ExtractionSource).
-
It declares port slots as annotated attributes whose type is another ABC
- (e.g. TransactionParserPort.number: NumberPort, amount_balance: AmountBalancePort).
-
It declares @abstractmethod bodies — the method signatures each concrete
- implementation must provide.
+
It declares attributes as class-level annotated defaults (e.g. TransactionParserPort.min_numbers: int = 3), optionally grouped as a nested struct via a @dataclass (e.g. source: ExtractionSource).
+
It declares port slots as annotated attributes whose type is another ABC (e.g. TransactionParserPort.number: NumberPort, amount_balance: AmountBalancePort).
+
It declares @abstractmethod bodies — the method signatures each concrete implementation must provide.
-
The ABC never runs, never holds an instance, and never says which implementation
-to use. It only says what the shape is.
The common default behavior ("Impl* — the shared adaptor").
-
class ImplTransactionParserPort(TransactionParserPort) subclasses the ABC,
- implements every @abstractmethod, and re-pins its port slots to concrete
- Impl* types: number: ImplNumberPort, desc: ImplDescPort,
- amount_balance: ImplAmountBalancePort.
-
ImplAmountBalancePort in turn pins number: ImplNumberPort and
- ambiguity: ImplAmbiguityPort.
-
The root, ImplExtractionPipeline, pins its five stage slots to Impl* ports:
- ingest, raw_lines, txn_blocks, txn_dicts, raw_expense.
+
class ImplTransactionParserPort(TransactionParserPort) subclasses the ABC, implements every @abstractmethod, and re-pins its port slots to concrete Impl* types: number: ImplNumberPort, desc: ImplDescPort, amount_balance: ImplAmountBalancePort.
+
ImplAmountBalancePort in turn pins number: ImplNumberPort and ambiguity: ImplAmbiguityPort.
+
The root, ImplExtractionPipeline, pins its five stage slots to Impl* ports: ingest, raw_lines, txn_blocks, txn_dicts, raw_expense.
-
Impl* is the default that shared, bank-agnostic behavior lives in. Any slot a
-bank does not override falls back to Impl* at runtime.
-
Layer 3 — bank specializations (samples/extraction_pipeline/banks/{axis,icici}/pdf.py, samples/extraction_pipeline/extraction_pipeline_banks.py)
+
Impl* is the default that shared, bank-agnostic behavior lives in. Any slot a bank does not override falls back to Impl* at runtime.
+
Layer 3 — Bank specializations (samples/extraction_pipeline/banks/{axis,icici}/pdf.py, samples/extraction_pipeline/extraction_pipeline_banks.py)¶
The variation ("what differs per bank").
-
Bank classes subclass the concrete Impl* classes, never the ABC — they
- extend, they never reinvent the contract.
-
Each Axis*/Icici* class overrides only the members that differ from the
- generic Impl*; everything shared is inherited.
-
A bank root is a thin annotated subclass of ImplExtractionPipeline that
- re-pins only the slots that differ:
+
Bank classes subclass the concrete Impl* classes, never the ABC — they extend, they never reinvent the contract.
+
Each Axis*/Icici* class overrides only the members that differ from the generic Impl*; everything shared is inherited.
+
A bank root is a thin annotated subclass of ImplExtractionPipeline that re-pins only the slots that differ:
Bank-specific wiring is hierarchical and cascading. Each level narrows exactly
-one slot, and its narrowed type drags the next level's narrowing along with it:
AxisExtractionPipeline
- └─ txn_dicts: AxisTxnDictsPort (only override on the root)
- └─ parser: AxisTransactionParserPort
- ├─ number: AxisNumberPort (get_numbers → all clean nums)
- ├─ desc: AxisDescPort (STARTTERS)
- └─ amount_balance: AxisAmountBalancePort
- └─ number: AxisNumberPort (re-pin: same bank's number port)
-
-
Siblings that are already correct in Impl* are inherited, not redeclared.
-E.g. the parser's date slot comes from ImplTransactionParserPort; the
-amount-balance port's ambiguity comes from ImplAmountBalancePort. Re-declaring
-them adds mental load on the extender with zero behavior change.
Bank-specific wiring is hierarchical and cascading. Each level narrows exactly one slot, and its narrowed type drags the next level's narrowing along with it:
+
Text Only
AxisExtractionPipeline
+ └─ txn_dicts: AxisTxnDictsPort (only override on the root)
+ └─ parser: AxisTransactionParserPort
+ ├─ number: AxisNumberPort (get_numbers → all clean nums)
+ ├─ desc: AxisDescPort (STARTTERS)
+ └─ amount_balance: AxisAmountBalancePort
+ └─ number: AxisNumberPort (re-pin: same bank's number port)
+
+
Siblings that are already correct in Impl* are inherited, not redeclared. E.g. the parser's date slot comes from ImplTransactionParserPort; the amount-balance port's ambiguity comes from ImplAmountBalancePort. Re-declaring them adds mental load on the extender with zero behavior change.
-
Rule — only redeclare the slot you are actually changing. If a slot's
-behavior is already correct from Impl*, leave it out.
+
Rule — only redeclare the slot you are actually changing. If a slot's behavior is already correct from Impl*, leave it out.
-
The two fidelity guarantees this pattern preserves
+
+
The Two Fidelity Guarantees This Pattern Preserves¶
-
Shared behavior stays shared. Slots like ingest, raw_lines, txn_blocks,
- raw_expense, parser date, and ambiguity are bank-agnostic. Leaving them at
- the Impl* default is deliberate, not an omission.
-
Banks never reinvent.Axis*/Icici* only extendImpl*. The ABC contract
- in samples/extraction_pipeline/extraction_pipeline_abc.py is defined once and shared by every bank.
+
Shared behavior stays shared. Slots like ingest, raw_lines, txn_blocks, raw_expense, parser date, and ambiguity are bank-agnostic. Leaving them at the Impl* default is deliberate, not an omission.
+
Banks never reinvent.Axis*/Icici* only extendImpl*. The ABC contract in samples/extraction_pipeline/extraction_pipeline_abc.py is defined once and shared by every bank.
No imperative assembly. No build_axis_pipeline() wiring objects by hand.
-
No constructor injection.Impl* classes have no __init__; wiring is done by
- assigning concrete port instances onto annotated slots.
-
YAML and ABCs are interchangeable, and both optional. The .yaml spec and
- the _abc.py module are two representations of the same contract. hexa is
- agnostic of the extraction pipeline sample and never imposes a workflow: a user can hand-write the
- ABC file, start from YAML (ABC generated), start from an ABC file (YAML
- emitted), or use both as a spec. The optional utilities in
- design.md § Utilities translate between them and can verify they
- agree — but the bank Impl*/specialization files are always hand-written
- (their slots are a manual choice) and never generated.
+
No constructor injection.Impl* classes have no __init__; wiring is done by assigning concrete port instances onto annotated slots.
+
YAML and ABCs are interchangeable, and both optional. The .yaml spec and the _abc.py module are two representations of the same contract. hexa is agnostic of the extraction pipeline sample and never imposes a workflow: a user can hand-write the ABC file, start from YAML (ABC generated), start from an ABC file (YAML emitted), or use both as a spec. The optional utilities in Design § Utilities translate between them and can verify they agree — but the bank Impl*/specialization files are always hand-written (their slots are a manual choice) and never generated.
The annotations are the decision; hexa.build(cls) (see Design § Container) turns a root class into a wired instance by reading __annotations__ + MRO, recursively instantiating each pinned concrete type, and setattring the child onto the parent slot. This document is scoped to the notation — the rest of the tree can be expressed purely as class annotations and filled at runtime by the container.
The annotations are the decision; hexa.build(cls) (see design.md §
-Container) turns a root class into a wired instance by reading __annotations__
-+ MRO, recursively instantiating each pinned concrete type, and setattring the
-child onto the parent slot. This document is scoped to the notation — the rest
-of the tree can be expressed purely as class annotations and filled at runtime by
-the container.