Abc

pipelines.enrichment.entity.abc

Summary

Abstract contracts for the entity enrichment family (no implementation).

---

The entity family is the classic dagpipe chain:

normalize → extract_candidates → candidate_flags → amount_flags →
set_entity → entity_alias → fallback

Each stage port is a dagpipe Node whose resolve yields forked States carrying confidence. EntityPipeline (the hexa root) wires the stage slots and runs the graph through a dagpipe Engine; EntityFinalizePort turns the terminal states into the entity result dict.

Classes

CandidateFlagsPort

Bases: RegistryPort, ABC

candidate.value → candidate.flags via CANDIDATE_BASED_FLAG_MAPPING.

Attributes

config property

config

Per-node config injected via set_config.

Returns:

Type	Description
	dict | None: The per-node config, or None before injection.

Functions

resolve abstractmethod

resolve(state: State) -> Iterable[State]

Yield forks with candidate.flags updated from the candidate value.

Parameters:

Name	Type	Description	Default
state	State	Incoming dagpipe state carrying candidate.	required

Yields:

Name	Type	Description
State	Iterable[State]	Forked states with candidate.flags populated.

set_config

set_config(config)

Inject the per-node config.

Parameters:

Name	Type	Description	Default
config	dict	Per-node config from the dagpipe spec.	required

set_registry

set_registry(registry)

Inject the merged pattern registry.

Parameters:

Name	Type	Description	Default
registry	PatternRegistry	The merged pattern registry.	required

EntityAliasPort

Bases: RegistryPort, ABC

candidate.value → entity via ENTITY_ALIAS_REGEX (when name not set).

Attributes

config property

config

Per-node config injected via set_config.

Returns:

Type	Description
	dict | None: The per-node config, or None before injection.

Functions

resolve abstractmethod

resolve(state: State) -> Iterable[State]

Yield the aliased entity fork from the candidate value.

Parameters:

Name	Type	Description	Default
state	State	Incoming dagpipe state carrying a candidate value.	required

Yields:

Name	Type	Description
State	Iterable[State]	Forked states with the aliased entity.

set_config

set_config(config)

Inject the per-node config.

Parameters:

Name	Type	Description	Default
config	dict	Per-node config from the dagpipe spec.	required

set_registry

set_registry(registry)

Inject the merged pattern registry.

Parameters:

Name	Type	Description	Default
registry	PatternRegistry	The merged pattern registry.	required

EntityFinalizePort

Bases: FinalizePort, ABC

terminal states → {entity.name, entity.id, entity.type, ...}.

Functions

process abstractmethod

process(states: List) -> dict

Build the entity result dict from the terminal states.

Parameters:

Name	Type	Description	Default
states	List	Terminal states from the dagpipe graph.	required

Returns:

Name	Type	Description
dict	dict	The entity result dict.

EntityPipeline

Bases: ABC

Root — wires normalize → … → fallback, then runs the dagpipe graph.

Attributes:

Name	Type	Description
normalize	NormalizePort	Normalize stage port.
extract_candidates	ExtractCandidatesPort	Extract-candidates stage port.
candidate_flags	CandidateFlagsPort	Candidate-flags stage port.
amount_flags	AmountFlagsPort	Amount-flags stage port.
set_entity	SetEntityPort	Set-entity stage port.
entity_alias	EntityAliasPort	Entity-alias stage port.
fallback	FallbackPort	Fallback stage port.
finalize	EntityFinalizePort	Terminal finalize port.

Functions

enrich abstractmethod

enrich(payload: dict) -> dict

Enrich a raw transaction payload into an entity result.

Parameters:

Name	Type	Description	Default
payload	dict	Raw transaction payload.	required

Returns:

Name	Type	Description
dict	dict	The entity result dict.

ExtractCandidatesPort

Bases: RegistryPort, ABC

signals → candidate.value forks (confidence penalties on fallbacks).

Attributes

config property

config

Per-node config injected via set_config.

Returns:

Type	Description
	dict | None: The per-node config, or None before injection.

Functions

resolve abstractmethod

resolve(state: State) -> Iterable[State]

Yield candidate.value forks, penalizing fallback candidates.

Parameters:

Name	Type	Description	Default
state	State	Incoming dagpipe state carrying signals.	required

Yields:

Name	Type	Description
State	Iterable[State]	Forked states carrying a candidate.value.

set_config

set_config(config)

Inject the per-node config.

Parameters:

Name	Type	Description	Default
config	dict	Per-node config from the dagpipe spec.	required

set_registry

set_registry(registry)

Inject the merged pattern registry.

Parameters:

Name	Type	Description	Default
registry	PatternRegistry	The merged pattern registry.	required

FallbackPort

Bases: RegistryPort, ABC

Terminal — merchant/person fallback or Unknown.

Attributes

config property

config

Per-node config injected via set_config.

Returns:

Type	Description
	dict | None: The per-node config, or None before injection.

Functions

resolve abstractmethod

resolve(state: State) -> Iterable[State]

Yield the terminal fallback entity fork.

Parameters:

Name	Type	Description	Default
state	State	Incoming terminal dagpipe state.	required

Yields:

Name	Type	Description
State	Iterable[State]	Forked states with the fallback entity.

set_config

set_config(config)

Inject the per-node config.

Parameters:

Name	Type	Description	Default
config	dict	Per-node config from the dagpipe spec.	required

set_registry

set_registry(registry)

Inject the merged pattern registry.

Parameters:

Name	Type	Description	Default
registry	PatternRegistry	The merged pattern registry.	required

NormalizePort

Bases: RegistryPort, ABC

raw.description → signals (raw/items/parts), stripping number sequences.

Attributes

config property

config

Per-node config injected via set_config.

Returns:

Type	Description
	dict | None: The per-node config, or None before injection.

Functions

resolve abstractmethod

resolve(state: State) -> Iterable[State]

Yield the normalized signals fork of the incoming state.

Parameters:

Name	Type	Description	Default
state	State	Incoming dagpipe state carrying raw.description.	required

Yields:

Name	Type	Description
State	Iterable[State]	Forked states carrying the normalized signals.

set_config

set_config(config)

Inject the per-node config.

Parameters:

Name	Type	Description	Default
config	dict	Per-node config from the dagpipe spec.	required

set_registry

set_registry(registry)

Inject the merged pattern registry.

Parameters:

Name	Type	Description	Default
registry	PatternRegistry	The merged pattern registry.	required

SetEntityPort

Bases: RegistryPort, ABC

candidate.flags → entity via FLAGS_BASED_ENTITY_RULES.

Attributes

config property

config

Per-node config injected via set_config.

Returns:

Type	Description
	dict | None: The per-node config, or None before injection.

Functions

resolve abstractmethod

resolve(state: State) -> Iterable[State]

Yield the resolved entity fork from candidate flags.

Parameters:

Name	Type	Description	Default
state	State	Incoming dagpipe state carrying candidate flags.	required

Yields:

Name	Type	Description
State	Iterable[State]	Forked states carrying the resolved entity.

set_config

set_config(config)

Inject the per-node config.

Parameters:

Name	Type	Description	Default
config	dict	Per-node config from the dagpipe spec.	required

set_registry

set_registry(registry)

Inject the merged pattern registry.

Parameters:

Name	Type	Description	Default
registry	PatternRegistry	The merged pattern registry.	required