Design — Extraction Pipeline (Axis / Icici)

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.

---

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.
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.

---

The port tree (ABC contract)

Text Only

ExtractionPipeline
├── source / pipeline / trust_fallback          [attributes; source is a nested struct]
├── ingest      → IngestPort                    (source → content)
├── raw_lines   → RawLinesPort                  (content → raw_lines)
├── txn_blocks  → TxnBlocksPort                 (raw_lines → txn_blocks)
│   ├── bucket_finder → BucketFinderPort
│   └── splitter      → SplitterPort
├── txn_dicts   → TxnDictsPort                  (txn_blocks → txn_dicts)
│   └── parser  → TransactionParserPort
│       ├── number         → NumberPort
│       ├── date           → DatePort
│       ├── desc           → DescPort
│       └── amount_balance → AmountBalancePort
│           ├── number    → NumberPort
│           └── ambiguity → AmbiguityPort
└── raw_expense → RawExpensePort                (txn_dicts → list[RawExpense], terminal)

---

Generic default (Impl*)

ImplExtractionPipeline pins every stage slot to a concrete Impl* port:

Text Only

ImplExtractionPipeline
├── ingest:      ImplIngestPort
├── raw_lines:   ImplRawLinesPort
├── txn_blocks:  ImplTxnBlocksPort
│   ├── bucket_finder: ImplBucketFinderPort
│   └── splitter:      ImplSplitterPort
├── txn_dicts:   ImplTxnDictsPort
│   └── parser:  ImplTransactionParserPort
│       ├── number:        ImplNumberPort
│       ├── date:          ImplDatePort
│       ├── desc:          ImplDescPort
│       └── amount_balance: ImplAmountBalancePort
│           ├── number:    ImplNumberPort
│           └── ambiguity: ImplAmbiguityPort
└── raw_expense: ImplRawExpensePort

---

Bank variation (only what differs)

The root class re-pins only txn_dicts. Everything else inherits Impl*.

Axis (banks/axis/pdf.py)

Python

class AxisExtractionPipeline(ImplExtractionPipeline):
    txn_dicts: AxisTxnDictsPort

Python

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.

Icici (banks/icici/pdf.py)

Python

class IciciExtractionPipeline(ImplExtractionPipeline):
    txn_dicts: IciciTxnDictsPort

Python

IciciTxnDictsPort(ImplTxnDictsPort)
  └─ parser: IciciTransactionParserPort(ImplTransactionParserPort)
      ├─ number:  IciciNumberPort         # NUMBER_RE, returns decimals[-2:] (last 2)
      ├─ desc:    IciciDescPort           # STARTTERS + fix_ocr (UPl→UPI, ..., ICIC]→ICICI)
      └─ amount_balance: IciciAmountBalancePort(ImplAmountBalancePort)
          ├─ number: IciciNumberPort      # re-pin to icici number port
          ├─ adjust_balance:              # neg-markers [(-), (-}, {-}, {-) , ()] → -abs(balance)
          └─ missing_number_candidates:   # Icici-only one-number hook

---

Redundancy rule (applied in both banks)

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:

Text Only

build(AxisExtractionPipeline)  →  wired AxisExtractionPipeline

Algorithm (build(cls)):

1. 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*).
2. For each port slot whose annotated type is a concrete Impl*/bank class (not an ABC, not a builtin), instantiate it.
3. Recurse into that child — repeat until leaves (classes that declare no further port slots).
4. setattr(parent, slot, child) to wire each slot onto the parent instance.
5. 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:

Text Only

run(source)
  ingest.materialize(source)            → content
  raw_lines.extract(content)            → raw_lines
  txn_blocks.extract(raw_lines)         → {opening_balance: [blocks]}
  txn_dicts.extract(txn_blocks)         → [txn dicts]
  raw_expense.build(txn_dicts)          → [RawExpense]   (terminal)

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.

Python

@dataclass
class PortNode:
    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.

4. generate_yaml(module_or_path, out_path=None) -> str

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).

Workflows (user's choice)

Text Only

# YAML-first
edit spec.yaml
  → generate_abc(spec.yaml, abc.py)          # regenerate ABCs
  → hand-write/update Impl* + banks
  → check_matches(spec.yaml, abc.py) == True # guard

# ABC-first
edit abc.py
  → generate_yaml(abc.py, spec.yaml)         # emit YAML spec
  → check_matches(spec.yaml, abc.py) == True

# hand-written both (no utilities at all)
edit abc.py + spec.yaml independently        # optional check_matches as a guard

Impl*/bank files are always hand-extended on top and are never generated.

---

Related

• 05 – Philosophy · 03 – Wiring & runtime · Overview