Impl

pipelines.enrichment.tags.impl

Summary

Concrete tags-family implementations (dagpipe stage nodes + root).

---

Stage bodies are ports of domain/enrichment/tags/flags.py. The root ImplTagsPipeline.enrich runs the hexa-wired stage graph through a dagpipe Engine and delegates terminal processing to ImplTagsFinalizePort (port of TagStatesProcessorAdaptor).

Classes

ImplEntityFlagsPort

Bases: EntityFlagsPort

Entity-flags stage: append entity_* flags from the entity name.

Attributes

config property

config

Per-node config injected via set_config.

Returns:

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

entity_based_flag_mapping property

entity_based_flag_mapping

Entity name → flag mapping used for entity flagging.

Returns:

Name	Type	Description
dict		The entity-based flag mapping.

Functions

resolve

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

Append matching entity_* flags and reset tags.

Parameters:

Name	Type	Description	Default
state	State	Incoming state carrying entity.name and candidate.flags.	required

Yields:

Name	Type	Description
State	Iterable[State]	A fork with updated candidate.flags and empty tags, or the unmodified state when no entity name is set.

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

ImplSetTagsPort

Bases: SetTagsPort

Set-tags stage: emit tags from the matching flag 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.

flags_based_tags_rules property

flags_based_tags_rules

Flag-matching rules that emit tags.

Returns:

Type	Description
	List[Dict]: The ordered tag rules.

Functions

resolve

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

Append tags emitted by the first matching flag rules.

Parameters:

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

Yields:

Name	Type	Description
State	Iterable[State]	A fork with tags updated when rules matched, else the unmodified state.

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

ImplTagsFinalizePort

Bases: TagsFinalizePort

Finalize terminal states into the tags result dict.

Functions

process

process(states: List) -> dict

Build the tags result dict, deduped and order-preserving.

Parameters:

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

Returns:

Name	Type	Description
dict	dict	{"tags": [...]} with unique tags.

ImplTagsPipeline

Bases: TagsPipeline

Concrete tags pipeline root wiring the stage ports.

Attributes:

Name	Type	Description
amount_flags	ImplAmountFlagsPort	Amount-flags stage port.
entity_flags	ImplEntityFlagsPort	Entity-flags stage port.
set_tags	ImplSetTagsPort	Set-tags stage port.
finalize	ImplTagsFinalizePort	Terminal finalize port.

Functions

enrich

enrich(payload: dict) -> dict

Run the tags graph on an entity payload and finalize the result.

Parameters:

Name	Type	Description	Default
payload	dict	Payload carrying entity.name and the amount.	required

Returns:

Name	Type	Description
dict	dict	The tags result dict from the finalize port.