Skip to content

graflo.architecture.graph_types

Graph runtime types and data structures (extraction, assembly, containers).

This package defines graph-processing data structures used across the ingestion pipeline and database adapters. It provides:

  • Core data types for vertices and edges
  • Database index configurations
  • Graph container implementations
  • Edge mapping and casting utilities
  • Action context for graph transformations

The package is designed to be database-agnostic, supporting both ArangoDB and Neo4j through the DBType enum. It provides a unified interface for working with graph data structures while allowing for database-specific optimizations and features.

Submodules (for lighter imports):

  • identifiers — vertex/edge id aliases and edge-key serialization
  • enums — EdgeMapping, EncodingType, IndexType, EdgeType, EdgeDirection, EdgeCastingType
  • index_config — ABCFields, Weight, Index
  • container — GraphContainer
  • location — LocationIndex, ProvenancePath
  • transform — TransformPayload, VertexRep, merge helpers
  • context — ExtractionContext, AssemblyContext, observations, ActionContext

Modules:

Name Description
container

Graph data container for vertices and edges.
context

Extraction and assembly runtime contexts.
edge_derivation

Edge-derivation wiring shared by contract (authoring) and pipeline (runtime).
enums

Graph runtime enumerations.
identifiers

Graph identifier aliases and edge-key serialization.
index_config

Index and weight configuration models.
location

Location indexing for nested graph traversal.
merge

Document fusion: several observations of one entity becoming one document.
transform

Transform payloads and observation merging.

Attributes

EdgeId = tuple[str, str, str | None] module-attribute

EdgePhysicalKey = tuple[str, str, str | None, str | None] module-attribute

GraphEntity = str | EdgeId module-attribute

VertexName = str module-attribute

__all__ = ['ABCFields', 'ActionContext', 'AssemblyContext', 'EdgeCastingType', 'EdgeDirection', 'EdgeId', 'EdgeIntent', 'EdgeMapping', 'EdgePhysicalKey', 'EdgeType', 'EncodingType', 'ExtractionContext', 'GraphAssemblyResult', 'GraphContainer', 'GraphEntity', 'Index', 'IndexType', 'ItemsView', 'LocationIndex', 'ProvenancePath', 'ResourceCastResult', 'TransformCastFailure', 'TransformObservation', 'TransformPayload', 'VertexName', 'VertexObservation', 'VertexRep', 'Weight', 'context_dict_from_transform_buffer_item', 'deserialize_edge_key', 'merge_observation_with_transform_buffer', 'serialize_edge_key'] module-attribute

Classes

ABCFields

Bases: ConfigBaseModel

Base model for entities that have fields.

Attributes:

Name Type Description
name str | None

Optional name of the entity
fields list[str]

List of field names
Source code in graflo/architecture/graph_types/index_config.py 
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
class ABCFields(ConfigBaseModel):
    """Base model for entities that have fields.

    Attributes:
        name: Optional name of the entity
        fields: List of field names
    """

    name: str | None = Field(
        default=None,
        description="Optional name of the entity (e.g. vertex name for composite field prefix).",
    )
    fields: list[str] = Field(
        default_factory=list,
        description="List of field names for this entity.",
    )
    keep_vertex_name: bool = Field(
        default=True,
        description="If True, composite field names use entity@field format; otherwise use field only.",
    )

    def cfield(self, x: str) -> str:
        """Creates a composite field name by combining the entity name with a field name.

        Args:
            x: Field name to combine with entity name

        Returns:
            Composite field name in format "entity@field"
        """
        return f"{self.name}@{x}" if self.keep_vertex_name else x

Attributes

fields = Field(default_factory=list, description='List of field names for this entity.') class-attribute instance-attribute
keep_vertex_name = Field(default=True, description='If True, composite field names use entity@field format; otherwise use field only.') class-attribute instance-attribute
name = Field(default=None, description='Optional name of the entity (e.g. vertex name for composite field prefix).') class-attribute instance-attribute

Methods:

cfield(x)

Creates a composite field name by combining the entity name with a field name.

Parameters:

Name Type Description Default
x str

Field name to combine with entity name

 required

Returns:

Type Description
str

Composite field name in format "entity@field"
Source code in graflo/architecture/graph_types/index_config.py 
33
34
35
36
37
38
39
40
41
42
def cfield(self, x: str) -> str:
    """Creates a composite field name by combining the entity name with a field name.

    Args:
        x: Field name to combine with entity name

    Returns:
        Composite field name in format "entity@field"
    """
    return f"{self.name}@{x}" if self.keep_vertex_name else x

ActionContext

Bases: ExtractionContext

Backward-compatible extraction+assembly context.

Kept for existing callers/tests while Wave 5 migrates call surfaces.

Source code in graflo/architecture/graph_types/context.py 
229
230
231
232
233
234
235
class ActionContext(ExtractionContext):
    """Backward-compatible extraction+assembly context.

    Kept for existing callers/tests while Wave 5 migrates call surfaces.
    """

    acc_global: Any = Field(default_factory=dd_factory)

Attributes

acc_global = Field(default_factory=dd_factory) class-attribute instance-attribute

AssemblyContext

Bases: ConfigBaseModel

Assembly-phase context built from extraction outputs.

Source code in graflo/architecture/graph_types/context.py 
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
class AssemblyContext(ConfigBaseModel):
    """Assembly-phase context built from extraction outputs."""

    model_config = ConfigDict(kw_only=True)  # ty: ignore[invalid-key]

    extraction: ExtractionContext
    acc_global: Any = Field(default_factory=dd_factory)

    @property
    def acc_vertex(self) -> Any:
        return self.extraction.acc_vertex

    @property
    def transform_buffer(self) -> Any:
        return self.extraction.transform_buffer

    @property
    def obs_buffer(self) -> Any:
        return self.extraction.obs_buffer

    @property
    def edge_intents(self) -> list[EdgeIntent]:
        return self.extraction.edge_intents

    @classmethod
    def from_extraction(cls, extraction: ExtractionContext) -> AssemblyContext:
        return cls(extraction=extraction)

Attributes

acc_global = Field(default_factory=dd_factory) class-attribute instance-attribute
acc_vertex property
edge_intents property
extraction instance-attribute
model_config = ConfigDict(kw_only=True) class-attribute instance-attribute
obs_buffer property
transform_buffer property

Methods:

from_extraction(extraction) classmethod
Source code in graflo/architecture/graph_types/context.py 
209
210
211
@classmethod
def from_extraction(cls, extraction: ExtractionContext) -> AssemblyContext:
    return cls(extraction=extraction)

EdgeCastingType

Bases: BaseEnum

Types of edge casting supported.

PAIR: Edges are cast as pairs of vertices PRODUCT: Edges are cast as combinations of vertex sets

Source code in graflo/architecture/graph_types/enums.py 
69
70
71
72
73
74
75
76
77
78
class EdgeCastingType(BaseEnum):
    """Types of edge casting supported.

    PAIR: Edges are cast as pairs of vertices
    PRODUCT: Edges are cast as combinations of vertex sets
    """

    PAIR = "pair"
    PRODUCT = "product"
    COMBINATIONS = "combinations"

Attributes

COMBINATIONS = 'combinations' class-attribute instance-attribute
PAIR = 'pair' class-attribute instance-attribute
PRODUCT = 'product' class-attribute instance-attribute

EdgeDirection

Bases: BaseEnum

Which orientations a read follows, relative to the anchor vertex.

OUT: anchor is the edge source (the historical, and still default, behaviour) IN: anchor is the edge target ANY: either orientation — the correct reading of a logically undirected edge

Not every backend can answer every value. Reverse reachability is free or cheap on most targets but is fixed at DDL time on TigerGraph; see :mod:graflo.db.edge_direction_support.

Source code in graflo/architecture/graph_types/enums.py 
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
class EdgeDirection(BaseEnum):
    """Which orientations a read follows, relative to the anchor vertex.

    OUT: anchor is the edge source (the historical, and still default, behaviour)
    IN: anchor is the edge target
    ANY: either orientation — the correct reading of a logically undirected edge

    Not every backend can answer every value. Reverse reachability is free or
    cheap on most targets but is fixed at DDL time on TigerGraph; see
    :mod:`graflo.db.edge_direction_support`.
    """

    OUT = "out"
    IN = "in"
    ANY = "any"

Attributes

ANY = 'any' class-attribute instance-attribute
IN = 'in' class-attribute instance-attribute
OUT = 'out' class-attribute instance-attribute

EdgeIntent

Bases: ConfigBaseModel

Typed edge assembly request emitted during extraction.

Source code in graflo/architecture/graph_types/context.py 
85
86
87
88
89
90
91
class EdgeIntent(ConfigBaseModel):
    """Typed edge assembly request emitted during extraction."""

    edge: Any
    location: LocationIndex | None = None
    provenance: ProvenancePath | None = None
    derivation: EdgeDerivation | None = None

Attributes

derivation = None class-attribute instance-attribute
edge instance-attribute
location = None class-attribute instance-attribute
provenance = None class-attribute instance-attribute

EdgeMapping

Bases: BaseEnum

Defines how edges are mapped between vertices.

ALL: Maps all vertices to all vertices ONE_N: Maps one vertex to many vertices

Source code in graflo/architecture/graph_types/enums.py 
 8
 9
10
11
12
13
14
15
16
class EdgeMapping(BaseEnum):
    """Defines how edges are mapped between vertices.

    ALL: Maps all vertices to all vertices
    ONE_N: Maps one vertex to many vertices
    """

    ALL = "all"
    ONE_N = "1-n"

Attributes

ALL = 'all' class-attribute instance-attribute
ONE_N = '1-n' class-attribute instance-attribute

EdgeType

Bases: BaseEnum

Defines how edges are handled in the graph database.

INDIRECT: Uses pre-existing DB structures and may be used after data ingestion DIRECT: Generated during ingestion from resource pipelines

Source code in graflo/architecture/graph_types/enums.py 
41
42
43
44
45
46
47
48
49
class EdgeType(BaseEnum):
    """Defines how edges are handled in the graph database.

    INDIRECT: Uses pre-existing DB structures and may be used after data ingestion
    DIRECT: Generated during ingestion from resource pipelines
    """

    INDIRECT = "indirect"
    DIRECT = "direct"

Attributes

DIRECT = 'direct' class-attribute instance-attribute
INDIRECT = 'indirect' class-attribute instance-attribute

EncodingType

Bases: BaseEnum

Supported character encodings for data input/output.

Source code in graflo/architecture/graph_types/enums.py 
19
20
21
22
23
class EncodingType(BaseEnum):
    """Supported character encodings for data input/output."""

    ISO_8859 = "ISO-8859-1"
    UTF_8 = "utf-8"

Attributes

ISO_8859 = 'ISO-8859-1' class-attribute instance-attribute
UTF_8 = 'utf-8' class-attribute instance-attribute

ExtractionContext

Bases: ConfigBaseModel

Extraction-phase context.

Attributes:

Name Type Description
acc_vertex Any

Local accumulation of extracted vertices
transform_buffer Any

Buffer for transform payloads (defaultdict[LocationIndex, list])
obs_buffer Any

Merged observation context per location (dict[LocationIndex, dict])
vertex_observations list[VertexObservation]

Explicit extracted vertex observations
transform_observations list[TransformObservation]

Explicit extracted transform observations
edge_intents list[EdgeIntent]

Explicit edge intents for assembly phase
Source code in graflo/architecture/graph_types/context.py 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
class ExtractionContext(ConfigBaseModel):
    """Extraction-phase context.

    Attributes:
        acc_vertex: Local accumulation of extracted vertices
        transform_buffer: Buffer for transform payloads (defaultdict[LocationIndex, list])
        obs_buffer: Merged observation context per location (dict[LocationIndex, dict])
        vertex_observations: Explicit extracted vertex observations
        transform_observations: Explicit extracted transform observations
        edge_intents: Explicit edge intents for assembly phase
    """

    model_config = ConfigDict(kw_only=True)  # ty: ignore[invalid-key]

    # Pydantic cannot schema nested defaultdict with custom key types (e.g. LocationIndex),
    # so we use Any; runtime type is as documented in Attributes
    acc_vertex: Any = Field(default_factory=outer_factory)
    transform_buffer: Any = Field(default_factory=_default_dict_transforms)
    obs_buffer: Any = Field(default_factory=_default_dict_observations)
    vertex_observations: list[VertexObservation] = Field(
        default_factory=_default_vertex_observations
    )
    transform_observations: list[TransformObservation] = Field(
        default_factory=_default_transform_observations
    )
    edge_intents: list[EdgeIntent] = Field(default_factory=_default_edge_intents)
    transform_failures: list[TransformCastFailure] = Field(
        default_factory=_default_transform_failures
    )

    def record_vertex_observation(
        self, *, vertex_name: str, location: LocationIndex, vertex: dict, ctx: dict
    ) -> None:
        self.vertex_observations.append(
            VertexObservation(
                vertex_name=vertex_name,
                location=location,
                vertex=vertex,
                ctx=ctx,
                provenance=ProvenancePath.from_lindex(location),
            )
        )

    def record_transform_observation(
        self, *, location: LocationIndex, payload: TransformPayload
    ) -> None:
        self.transform_observations.append(
            TransformObservation(
                location=location,
                payload=payload,
                provenance=ProvenancePath.from_lindex(location),
            )
        )

    def record_edge_intent(
        self,
        *,
        edge: Any,
        location: LocationIndex,
        derivation: EdgeDerivation | None = None,
    ) -> None:
        self.edge_intents.append(
            EdgeIntent(
                edge=edge,
                location=location,
                provenance=ProvenancePath.from_lindex(location),
                derivation=derivation,
            )
        )

    def record_transform_failure(
        self,
        *,
        location: LocationIndex,
        transform_label: str,
        exc: BaseException,
        traceback_text: str,
        nulled_fields: tuple[str, ...],
    ) -> None:
        self.transform_failures.append(
            TransformCastFailure(
                location=location,
                transform_label=transform_label,
                exception_type=type(exc).__name__,
                message=str(exc),
                traceback=traceback_text,
                nulled_fields=nulled_fields,
            )
        )

Attributes

acc_vertex = Field(default_factory=outer_factory) class-attribute instance-attribute
edge_intents = Field(default_factory=_default_edge_intents) class-attribute instance-attribute
model_config = ConfigDict(kw_only=True) class-attribute instance-attribute
obs_buffer = Field(default_factory=_default_dict_observations) class-attribute instance-attribute
transform_buffer = Field(default_factory=_default_dict_transforms) class-attribute instance-attribute
transform_failures = Field(default_factory=_default_transform_failures) class-attribute instance-attribute
transform_observations = Field(default_factory=_default_transform_observations) class-attribute instance-attribute
vertex_observations = Field(default_factory=_default_vertex_observations) class-attribute instance-attribute

Methods:

record_edge_intent(*, edge, location, derivation=None)
Source code in graflo/architecture/graph_types/context.py 
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
def record_edge_intent(
    self,
    *,
    edge: Any,
    location: LocationIndex,
    derivation: EdgeDerivation | None = None,
) -> None:
    self.edge_intents.append(
        EdgeIntent(
            edge=edge,
            location=location,
            provenance=ProvenancePath.from_lindex(location),
            derivation=derivation,
        )
    )
record_transform_failure(*, location, transform_label, exc, traceback_text, nulled_fields)
Source code in graflo/architecture/graph_types/context.py 
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
def record_transform_failure(
    self,
    *,
    location: LocationIndex,
    transform_label: str,
    exc: BaseException,
    traceback_text: str,
    nulled_fields: tuple[str, ...],
) -> None:
    self.transform_failures.append(
        TransformCastFailure(
            location=location,
            transform_label=transform_label,
            exception_type=type(exc).__name__,
            message=str(exc),
            traceback=traceback_text,
            nulled_fields=nulled_fields,
        )
    )
record_transform_observation(*, location, payload)
Source code in graflo/architecture/graph_types/context.py 
137
138
139
140
141
142
143
144
145
146
def record_transform_observation(
    self, *, location: LocationIndex, payload: TransformPayload
) -> None:
    self.transform_observations.append(
        TransformObservation(
            location=location,
            payload=payload,
            provenance=ProvenancePath.from_lindex(location),
        )
    )
record_vertex_observation(*, vertex_name, location, vertex, ctx)
Source code in graflo/architecture/graph_types/context.py 
124
125
126
127
128
129
130
131
132
133
134
135
def record_vertex_observation(
    self, *, vertex_name: str, location: LocationIndex, vertex: dict, ctx: dict
) -> None:
    self.vertex_observations.append(
        VertexObservation(
            vertex_name=vertex_name,
            location=location,
            vertex=vertex,
            ctx=ctx,
            provenance=ProvenancePath.from_lindex(location),
        )
    )

GraphAssemblyResult

Bases: ConfigBaseModel

Result of graph assembly phase.

Source code in graflo/architecture/graph_types/context.py 
214
215
216
217
class GraphAssemblyResult(ConfigBaseModel):
    """Result of graph assembly phase."""

    entities: Any = Field(default_factory=dd_factory)

Attributes

entities = Field(default_factory=dd_factory) class-attribute instance-attribute

GraphContainer

Bases: ConfigBaseModel

Container for graph data including vertices and edges.

Attributes:

Name Type Description
vertices dict[VertexName, list]

Dictionary mapping vertex names to lists of vertex data
edges dict[tuple[str, str, str | None], list]

Dictionary mapping edge IDs to lists of edge data
linear list[defaultdict[str | tuple[str, str, str | None], list[Any]]]

List of default dictionaries containing linear data
Source code in graflo/architecture/graph_types/container.py 
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
class GraphContainer(ConfigBaseModel):
    """Container for graph data including vertices and edges.

    Attributes:
        vertices: Dictionary mapping vertex names to lists of vertex data
        edges: Dictionary mapping edge IDs to lists of edge data
        linear: List of default dictionaries containing linear data
    """

    vertices: dict[VertexName, list] = Field(default_factory=dict)
    edges: dict[tuple[str, str, str | None], list] = Field(default_factory=dict)
    linear: list[defaultdict[str | tuple[str, str, str | None], list[Any]]] = Field(
        default_factory=list
    )

    @field_serializer("edges", when_used="json-unless-none")
    def _serialize_edges(
        self, edges: dict[tuple[str, str, str | None], list]
    ) -> dict[str, list]:
        return {serialize_edge_key(k): v for k, v in edges.items()}

    @field_validator("edges", mode="before")
    @classmethod
    def _validate_edges(cls, value: Any) -> Any:
        if not isinstance(value, dict):
            return value
        normalized: dict[tuple[str, str, str | None], list] = {}
        for key, docs in value.items():
            if isinstance(key, str):
                normalized[deserialize_edge_key(key)] = docs
            else:
                normalized[key] = docs
        return normalized

    @field_serializer("linear", when_used="json-unless-none")
    def _serialize_linear(
        self,
        linear: list[defaultdict[str | tuple[str, str, str | None], list[Any]]],
    ) -> list[dict[str, list[Any]]]:
        return [_serialize_linear_item(item) for item in linear]

    @field_validator("linear", mode="before")
    @classmethod
    def _validate_linear(cls, value: Any) -> Any:
        if not isinstance(value, list):
            return value
        return [
            _deserialize_linear_item(item) if isinstance(item, dict) else item
            for item in value
        ]

    def items(self):
        """Get an ItemsView of the container's contents."""
        return ItemsView(self)

    def pick_unique(self):
        """Remove duplicate entries from vertices and edges."""
        for k, v in self.vertices.items():
            self.vertices[k] = pick_unique_dict(v)
        for k, v in self.edges.items():
            self.edges[k] = pick_unique_dict(v)

    @classmethod
    def from_docs_list(
        cls, list_default_dicts: list[defaultdict[GraphEntity, list]]
    ) -> GraphContainer:
        """Create a GraphContainer from a list of default dictionaries.

        Args:
            list_default_dicts: List of default dictionaries containing vertex and edge data

        Returns:
            New GraphContainer instance

        Raises:
            ValueError: If edge IDs are not properly formatted
        """
        vdict: defaultdict[str, list] = defaultdict(list)
        edict: defaultdict[tuple[str, str, str | None], list] = defaultdict(list)

        for d in list_default_dicts:
            for k, v in d.items():
                if isinstance(k, str):
                    vdict[k].extend(v)
                elif isinstance(k, tuple):
                    if not (
                        len(k) == 3
                        and all(isinstance(item, str) for item in k[:-1])
                        and isinstance(k[-1], (str, type(None)))
                    ):
                        raise ValueError(
                            f"edge key must be (str, str, str|None), got {k}"
                        )
                    edict[k].extend(v)
        return GraphContainer(
            vertices=dict(vdict.items()),
            edges=dict(edict.items()),
            linear=list_default_dicts,
        )

Attributes

edges = Field(default_factory=dict) class-attribute instance-attribute
linear = Field(default_factory=list) class-attribute instance-attribute
vertices = Field(default_factory=dict) class-attribute instance-attribute

Methods:

from_docs_list(list_default_dicts) classmethod

Create a GraphContainer from a list of default dictionaries.

Parameters:

Name Type Description Default
list_default_dicts list[defaultdict[GraphEntity, list]]

List of default dictionaries containing vertex and edge data

 required

Returns:

Type Description
GraphContainer

New GraphContainer instance

Raises:

Type Description
ValueError

If edge IDs are not properly formatted
Source code in graflo/architecture/graph_types/container.py 
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
@classmethod
def from_docs_list(
    cls, list_default_dicts: list[defaultdict[GraphEntity, list]]
) -> GraphContainer:
    """Create a GraphContainer from a list of default dictionaries.

    Args:
        list_default_dicts: List of default dictionaries containing vertex and edge data

    Returns:
        New GraphContainer instance

    Raises:
        ValueError: If edge IDs are not properly formatted
    """
    vdict: defaultdict[str, list] = defaultdict(list)
    edict: defaultdict[tuple[str, str, str | None], list] = defaultdict(list)

    for d in list_default_dicts:
        for k, v in d.items():
            if isinstance(k, str):
                vdict[k].extend(v)
            elif isinstance(k, tuple):
                if not (
                    len(k) == 3
                    and all(isinstance(item, str) for item in k[:-1])
                    and isinstance(k[-1], (str, type(None)))
                ):
                    raise ValueError(
                        f"edge key must be (str, str, str|None), got {k}"
                    )
                edict[k].extend(v)
    return GraphContainer(
        vertices=dict(vdict.items()),
        edges=dict(edict.items()),
        linear=list_default_dicts,
    )
items()

Get an ItemsView of the container's contents.

Source code in graflo/architecture/graph_types/container.py 
104
105
106
def items(self):
    """Get an ItemsView of the container's contents."""
    return ItemsView(self)
pick_unique()

Remove duplicate entries from vertices and edges.

Source code in graflo/architecture/graph_types/container.py 
108
109
110
111
112
113
def pick_unique(self):
    """Remove duplicate entries from vertices and edges."""
    for k, v in self.vertices.items():
        self.vertices[k] = pick_unique_dict(v)
    for k, v in self.edges.items():
        self.edges[k] = pick_unique_dict(v)

Index

Bases: ConfigBaseModel

Configuration for database indexes.

Attributes:

Name Type Description
name str | None

Optional name of the index
fields list[str]

List of fields to index
unique bool

Whether the index enforces uniqueness
type IndexType

Type of index to create
deduplicate bool

Whether to deduplicate index entries
sparse bool

Whether to create a sparse index
exclude_edge_endpoints bool

Whether to exclude edge endpoints from index
Source code in graflo/architecture/graph_types/index_config.py 
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
class Index(ConfigBaseModel):
    """Configuration for database indexes.

    Attributes:
        name: Optional name of the index
        fields: List of fields to index
        unique: Whether the index enforces uniqueness
        type: Type of index to create
        deduplicate: Whether to deduplicate index entries
        sparse: Whether to create a sparse index
        exclude_edge_endpoints: Whether to exclude edge endpoints from index
    """

    name: str | None = Field(
        default=None,
        description="Optional index name. For edges, can reference a vertex name for composite fields.",
    )
    fields: list[str] = Field(
        default_factory=list,
        description="List of field names included in this index.",
    )
    unique: bool = Field(
        default=True,
        description="If True, index enforces uniqueness on the field combination.",
    )
    type: IndexType = Field(
        default=IndexType.PERSISTENT,
        description="Index type (PERSISTENT, HASH, SKIPLIST, FULLTEXT).",
    )
    deduplicate: bool = Field(
        default=True,
        description="Whether to deduplicate index entries (e.g. ArangoDB).",
    )
    sparse: bool = Field(
        default=False,
        description="If True, create a sparse index (exclude null/missing values).",
    )
    exclude_edge_endpoints: bool = Field(
        default=False,
        description="If True, do not add _from/_to to edge index (e.g. ArangoDB).",
    )

    def __iter__(self):
        """Iterate over the indexed fields."""
        return iter(self.fields)

    def db_form(self, db_type: DBType) -> dict:
        """Convert index configuration to database-specific format.

        Args:
            db_type: Type of database (ARANGO or NEO4J)

        Returns:
            Dictionary of index configuration in database-specific format

        Raises:
            ValueError: If db_type is not supported
        """
        r = dict(self.to_dict())
        if db_type == DBType.ARANGO:
            r.pop("name", None)
            r.pop("exclude_edge_endpoints", None)
        return r

Attributes

deduplicate = Field(default=True, description='Whether to deduplicate index entries (e.g. ArangoDB).') class-attribute instance-attribute
exclude_edge_endpoints = Field(default=False, description='If True, do not add _from/_to to edge index (e.g. ArangoDB).') class-attribute instance-attribute
fields = Field(default_factory=list, description='List of field names included in this index.') class-attribute instance-attribute
name = Field(default=None, description='Optional index name. For edges, can reference a vertex name for composite fields.') class-attribute instance-attribute
sparse = Field(default=False, description='If True, create a sparse index (exclude null/missing values).') class-attribute instance-attribute
type = Field(default=IndexType.PERSISTENT, description='Index type (PERSISTENT, HASH, SKIPLIST, FULLTEXT).') class-attribute instance-attribute
unique = Field(default=True, description='If True, index enforces uniqueness on the field combination.') class-attribute instance-attribute

Methods:

__iter__()

Iterate over the indexed fields.

Source code in graflo/architecture/graph_types/index_config.py 
105
106
107
def __iter__(self):
    """Iterate over the indexed fields."""
    return iter(self.fields)
db_form(db_type)

Convert index configuration to database-specific format.

Parameters:

Name Type Description Default
db_type DBType

Type of database (ARANGO or NEO4J)

 required

Returns:

Type Description
dict

Dictionary of index configuration in database-specific format

Raises:

Type Description
ValueError

If db_type is not supported
Source code in graflo/architecture/graph_types/index_config.py 
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
def db_form(self, db_type: DBType) -> dict:
    """Convert index configuration to database-specific format.

    Args:
        db_type: Type of database (ARANGO or NEO4J)

    Returns:
        Dictionary of index configuration in database-specific format

    Raises:
        ValueError: If db_type is not supported
    """
    r = dict(self.to_dict())
    if db_type == DBType.ARANGO:
        r.pop("name", None)
        r.pop("exclude_edge_endpoints", None)
    return r

IndexType

Bases: BaseEnum

Types of database indexes supported.

PERSISTENT: Standard persistent index HASH: Hash-based index for fast lookups SKIPLIST: Sorted index using skip list data structure FULLTEXT: Index optimized for text search

Source code in graflo/architecture/graph_types/enums.py 
26
27
28
29
30
31
32
33
34
35
36
37
38
class IndexType(BaseEnum):
    """Types of database indexes supported.

    PERSISTENT: Standard persistent index
    HASH: Hash-based index for fast lookups
    SKIPLIST: Sorted index using skip list data structure
    FULLTEXT: Index optimized for text search
    """

    PERSISTENT = "persistent"
    HASH = "hash"
    SKIPLIST = "skiplist"
    FULLTEXT = "fulltext"

Attributes

FULLTEXT = 'fulltext' class-attribute instance-attribute
HASH = 'hash' class-attribute instance-attribute
PERSISTENT = 'persistent' class-attribute instance-attribute
SKIPLIST = 'skiplist' class-attribute instance-attribute

ItemsView

View class for iterating over vertices and edges in a GraphContainer.

Source code in graflo/architecture/graph_types/container.py 
39
40
41
42
43
44
45
46
47
48
49
50
class ItemsView:
    """View class for iterating over vertices and edges in a GraphContainer."""

    def __init__(self, gc: GraphContainer):
        self._dictlike = gc

    def __iter__(self):
        """Iterate over vertices and edges in the container."""
        for key in self._dictlike.vertices:
            yield key, self._dictlike.vertices[key]
        for key in self._dictlike.edges:
            yield key, self._dictlike.edges[key]

Methods:

__init__(gc)
Source code in graflo/architecture/graph_types/container.py 
42
43
def __init__(self, gc: GraphContainer):
    self._dictlike = gc
__iter__()

Iterate over vertices and edges in the container.

Source code in graflo/architecture/graph_types/container.py 
45
46
47
48
49
50
def __iter__(self):
    """Iterate over vertices and edges in the container."""
    for key in self._dictlike.vertices:
        yield key, self._dictlike.vertices[key]
    for key in self._dictlike.edges:
        yield key, self._dictlike.edges[key]

LocationIndex

Bases: ConfigBaseModel

Immutable location index for nested graph traversal.

Source code in graflo/architecture/graph_types/location.py 
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
class LocationIndex(ConfigBaseModel):
    """Immutable location index for nested graph traversal."""

    model_config = ConfigDict(frozen=True)

    path: tuple[str | int | None, ...] = Field(default_factory=tuple)

    @model_validator(mode="before")
    @classmethod
    def accept_tuple(cls, data: Any) -> Any:
        """Accept a single tuple as positional path (e.g. LocationIndex((0,)))."""
        if isinstance(data, tuple):
            return {"path": data}
        return data

    def __init__(self, *args: Any, **kwargs: Any) -> None:
        """Allow LocationIndex((0,)) or LocationIndex(path=(0,))."""
        if args and len(args) == 1 and isinstance(args[0], tuple) and not kwargs:
            kwargs = {"path": args[0]}
        super().__init__(**kwargs)

    def extend(self, extension: tuple[str | int | None, ...]) -> LocationIndex:
        return LocationIndex(path=(*self.path, *extension))

    def parent(self) -> LocationIndex | None:
        if not self.path:
            return None
        return LocationIndex(path=self.path[:-1])

    def depth(self) -> int:
        return len(self.path)

    def congruence_measure(self, other: LocationIndex) -> int:
        neq_position = 0
        for step_a, step_b in zip(self.path, other.path):
            if step_a != step_b:
                break
            neq_position += 1
        return neq_position

    def filter(self, lindex_list: list[LocationIndex]) -> list[LocationIndex]:
        return [
            t
            for t in lindex_list
            if t.depth() >= self.depth() and t.path[: self.depth()] == self.path
        ]

    def __lt__(self, other: LocationIndex) -> bool:
        return len(self.path) < len(other.path)

    def __contains__(self, item: object) -> bool:
        return item in self.path

    def __len__(self) -> int:
        return len(self.path)

    def __iter__(self):
        return iter(self.path)

    def __getitem__(self, item: int | slice):
        return self.path[item]

Attributes

model_config = ConfigDict(frozen=True) class-attribute instance-attribute
path = Field(default_factory=tuple) class-attribute instance-attribute

Methods:

__contains__(item)
Source code in graflo/architecture/graph_types/location.py 
62
63
def __contains__(self, item: object) -> bool:
    return item in self.path
__getitem__(item)
Source code in graflo/architecture/graph_types/location.py 
71
72
def __getitem__(self, item: int | slice):
    return self.path[item]
__init__(*args, **kwargs)

Allow LocationIndex((0,)) or LocationIndex(path=(0,)).

Source code in graflo/architecture/graph_types/location.py 
27
28
29
30
31
def __init__(self, *args: Any, **kwargs: Any) -> None:
    """Allow LocationIndex((0,)) or LocationIndex(path=(0,))."""
    if args and len(args) == 1 and isinstance(args[0], tuple) and not kwargs:
        kwargs = {"path": args[0]}
    super().__init__(**kwargs)
__iter__()
Source code in graflo/architecture/graph_types/location.py 
68
69
def __iter__(self):
    return iter(self.path)
__len__()
Source code in graflo/architecture/graph_types/location.py 
65
66
def __len__(self) -> int:
    return len(self.path)
__lt__(other)
Source code in graflo/architecture/graph_types/location.py 
59
60
def __lt__(self, other: LocationIndex) -> bool:
    return len(self.path) < len(other.path)
accept_tuple(data) classmethod

Accept a single tuple as positional path (e.g. LocationIndex((0,))).

Source code in graflo/architecture/graph_types/location.py 
19
20
21
22
23
24
25
@model_validator(mode="before")
@classmethod
def accept_tuple(cls, data: Any) -> Any:
    """Accept a single tuple as positional path (e.g. LocationIndex((0,)))."""
    if isinstance(data, tuple):
        return {"path": data}
    return data
congruence_measure(other)
Source code in graflo/architecture/graph_types/location.py 
44
45
46
47
48
49
50
def congruence_measure(self, other: LocationIndex) -> int:
    neq_position = 0
    for step_a, step_b in zip(self.path, other.path):
        if step_a != step_b:
            break
        neq_position += 1
    return neq_position
depth()
Source code in graflo/architecture/graph_types/location.py 
41
42
def depth(self) -> int:
    return len(self.path)
extend(extension)
Source code in graflo/architecture/graph_types/location.py 
33
34
def extend(self, extension: tuple[str | int | None, ...]) -> LocationIndex:
    return LocationIndex(path=(*self.path, *extension))
filter(lindex_list)
Source code in graflo/architecture/graph_types/location.py 
52
53
54
55
56
57
def filter(self, lindex_list: list[LocationIndex]) -> list[LocationIndex]:
    return [
        t
        for t in lindex_list
        if t.depth() >= self.depth() and t.path[: self.depth()] == self.path
    ]
parent()
Source code in graflo/architecture/graph_types/location.py 
36
37
38
39
def parent(self) -> LocationIndex | None:
    if not self.path:
        return None
    return LocationIndex(path=self.path[:-1])

ProvenancePath

Bases: ConfigBaseModel

Explicit provenance path for extracted observations.

Source code in graflo/architecture/graph_types/location.py 
75
76
77
78
79
80
81
82
class ProvenancePath(ConfigBaseModel):
    """Explicit provenance path for extracted observations."""

    path: tuple[str | int | None, ...] = Field(default_factory=tuple)

    @classmethod
    def from_lindex(cls, lindex: LocationIndex) -> ProvenancePath:
        return cls(path=lindex.path)

Attributes

path = Field(default_factory=tuple) class-attribute instance-attribute

Methods:

from_lindex(lindex) classmethod
Source code in graflo/architecture/graph_types/location.py 
80
81
82
@classmethod
def from_lindex(cls, lindex: LocationIndex) -> ProvenancePath:
    return cls(path=lindex.path)

ResourceCastResult

Bases: ConfigBaseModel

Outcome of casting one document through a resource pipeline.

Source code in graflo/architecture/graph_types/context.py 
220
221
222
223
224
225
226
class ResourceCastResult(ConfigBaseModel):
    """Outcome of casting one document through a resource pipeline."""

    model_config = ConfigDict(arbitrary_types_allowed=True)

    entities: Any
    transform_failures: list[TransformCastFailure] = Field(default_factory=list)

Attributes

entities instance-attribute
model_config = ConfigDict(arbitrary_types_allowed=True) class-attribute instance-attribute
transform_failures = Field(default_factory=list) class-attribute instance-attribute

TransformCastFailure

Bases: ConfigBaseModel

One transform step that failed during extraction (tolerance mode).

Source code in graflo/architecture/graph_types/context.py 
74
75
76
77
78
79
80
81
82
class TransformCastFailure(ConfigBaseModel):
    """One transform step that failed during extraction (tolerance mode)."""

    location: LocationIndex
    transform_label: str
    exception_type: str
    message: str
    traceback: str = ""
    nulled_fields: tuple[str, ...] = Field(default_factory=tuple)

Attributes

exception_type instance-attribute
location instance-attribute
message instance-attribute
nulled_fields = Field(default_factory=tuple) class-attribute instance-attribute
traceback = '' class-attribute instance-attribute
transform_label instance-attribute

TransformObservation

Bases: ConfigBaseModel

Typed transform observation emitted during extraction.

Source code in graflo/architecture/graph_types/context.py 
66
67
68
69
70
71
class TransformObservation(ConfigBaseModel):
    """Typed transform observation emitted during extraction."""

    location: LocationIndex
    payload: TransformPayload
    provenance: ProvenancePath

Attributes

location instance-attribute
payload instance-attribute
provenance instance-attribute

TransformPayload

Bases: ConfigBaseModel

Typed transform output shared between extraction and assembly phases.

Source code in graflo/architecture/graph_types/transform.py 
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
class TransformPayload(ConfigBaseModel):
    """Typed transform output shared between extraction and assembly phases."""

    named: dict[str, Any] = Field(default_factory=dict)
    positional: tuple[Any, ...] = Field(default_factory=tuple)
    removed_keys: frozenset[str] = Field(default_factory=frozenset)

    @classmethod
    def from_result(cls, result: Any) -> TransformPayload:
        if isinstance(result, dict):
            return cls(named=dict(result))
        if isinstance(result, tuple):
            return cls(positional=tuple(result))
        return cls(positional=(result,))

    def context_doc(self) -> dict[str, Any]:
        """Named values available for weight and relation extraction."""
        return dict(self.named)

Attributes

named = Field(default_factory=dict) class-attribute instance-attribute
positional = Field(default_factory=tuple) class-attribute instance-attribute
removed_keys = Field(default_factory=frozenset) class-attribute instance-attribute

Methods:

context_doc()

Named values available for weight and relation extraction.

Source code in graflo/architecture/graph_types/transform.py 
46
47
48
def context_doc(self) -> dict[str, Any]:
    """Named values available for weight and relation extraction."""
    return dict(self.named)
from_result(result) classmethod
Source code in graflo/architecture/graph_types/transform.py 
38
39
40
41
42
43
44
@classmethod
def from_result(cls, result: Any) -> TransformPayload:
    if isinstance(result, dict):
        return cls(named=dict(result))
    if isinstance(result, tuple):
        return cls(positional=tuple(result))
    return cls(positional=(result,))

VertexObservation

Bases: ConfigBaseModel

Typed vertex observation emitted during extraction.

Source code in graflo/architecture/graph_types/context.py 
56
57
58
59
60
61
62
63
class VertexObservation(ConfigBaseModel):
    """Typed vertex observation emitted during extraction."""

    vertex_name: str
    location: LocationIndex
    vertex: dict[str, Any]
    ctx: dict[str, Any]
    provenance: ProvenancePath

Attributes

ctx instance-attribute
location instance-attribute
provenance instance-attribute
vertex instance-attribute
vertex_name instance-attribute

VertexRep

Bases: ConfigBaseModel

Context for graph transformation actions.

Attributes:

Name Type Description
vertex dict[str, Any]

doc representing a vertex
Source code in graflo/architecture/graph_types/transform.py 
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
class VertexRep(ConfigBaseModel):
    """Context for graph transformation actions.

    Attributes:
        vertex: doc representing a vertex
    """

    model_config = ConfigDict(kw_only=True)  # ty: ignore[invalid-key]

    vertex: dict[str, Any]
    lookup_only: bool = False
    """True when this observation exists only to locate an existing vertex.

    Such documents take part in edge rendering but are never upserted. Tagging
    the observation rather than the vertex type keeps it correct when one
    resource both writes and merely references the same vertex type.
    """

Attributes

lookup_only = False class-attribute instance-attribute

True when this observation exists only to locate an existing vertex.

Such documents take part in edge rendering but are never upserted. Tagging the observation rather than the vertex type keeps it correct when one resource both writes and merely references the same vertex type.

model_config = ConfigDict(kw_only=True) class-attribute instance-attribute
vertex instance-attribute

Weight

Bases: ABCFields

Defines weight configuration for edges.

Attributes:

Name Type Description
map dict

Dictionary mapping field values to weights
filter dict

Dictionary of filter conditions for weights
Source code in graflo/architecture/graph_types/index_config.py 
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
class Weight(ABCFields):
    """Defines weight configuration for edges.

    Attributes:
        map: Dictionary mapping field values to weights
        filter: Dictionary of filter conditions for weights
    """

    map: dict = Field(
        default_factory=dict,
        description="Mapping of field values to weight values for vertex-based edge attributes.",
    )
    filter: dict = Field(
        default_factory=dict,
        description="Filter conditions applied when resolving vertex-based weights.",
    )

Attributes

filter = Field(default_factory=dict, description='Filter conditions applied when resolving vertex-based weights.') class-attribute instance-attribute
map = Field(default_factory=dict, description='Mapping of field values to weight values for vertex-based edge attributes.') class-attribute instance-attribute

Functions:

context_dict_from_transform_buffer_item(item)

Map one transform_buffer entry to a flat context dict (named keys only).

Source code in graflo/architecture/graph_types/transform.py 
51
52
53
54
55
56
57
def context_dict_from_transform_buffer_item(item: Any) -> dict[str, Any]:
    """Map one ``transform_buffer`` entry to a flat context dict (named keys only)."""
    if isinstance(item, TransformPayload):
        return item.context_doc()
    if isinstance(item, dict):
        return dict(item)
    return {}

deserialize_edge_key(key)

Deserialize a JSON-array edge key back to an edge id tuple.

Source code in graflo/architecture/graph_types/identifiers.py 
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
def deserialize_edge_key(key: str) -> tuple[str, str, str | None]:
    """Deserialize a JSON-array edge key back to an edge id tuple."""
    try:
        parsed = json.loads(key)
    except json.JSONDecodeError as exc:
        raise ValueError(f"Invalid edge key JSON: {key!r}") from exc
    if not isinstance(parsed, list) or len(parsed) != 3:
        raise ValueError(f"Invalid edge key JSON array: {key!r}")
    if not all(isinstance(part, str) for part in parsed[:2]):
        raise ValueError(f"Invalid edge key JSON array: {key!r}")
    relation_part = parsed[2]
    if relation_part is not None and not isinstance(relation_part, str):
        raise ValueError(f"Invalid edge key JSON array: {key!r}")
    relation = relation_part if relation_part else None
    return (parsed[0], parsed[1], relation)

merge_observation_with_transform_buffer(observation, buffer_items)

Merge a JSON observation slice with transform outputs at the same location.

observation is the current dict-shaped fragment of the nested document passed into actors (often a child object under a :class:DescendActor). buffer_items are the entries in ExtractionContext.transform_buffer for the same :class:LocationIndex.

Starts from a shallow copy of observation; each buffer entry (in pipeline order) updates the merged view, so later transforms override earlier keys and transform output overrides the raw JSON on key conflicts.

Source code in graflo/architecture/graph_types/transform.py 
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
def merge_observation_with_transform_buffer(
    observation: dict[str, Any],
    buffer_items: list[Any],
) -> dict[str, Any]:
    """Merge a JSON observation slice with transform outputs at the same location.

    ``observation`` is the current dict-shaped fragment of the nested document
    passed into actors (often a child object under a :class:`DescendActor`).
    ``buffer_items`` are the entries in ``ExtractionContext.transform_buffer``
    for the same :class:`LocationIndex`.

    Starts from a shallow copy of ``observation``; each buffer entry (in pipeline
    order) updates the merged view, so later transforms override earlier keys
    and transform output overrides the raw JSON on key conflicts.
    """
    merged: dict[str, Any] = dict(observation)
    for item in buffer_items:
        merged.update(context_dict_from_transform_buffer_item(item))
        if isinstance(item, TransformPayload) and item.removed_keys:
            for k in item.removed_keys:
                merged.pop(k, None)
    return merged

serialize_edge_key(edge_id)

Serialize an edge id tuple to a JSON-safe string key.

Uses a JSON array [source, target, relation] so vertex or relation names may contain | or other special characters without ambiguity.

Source code in graflo/architecture/graph_types/identifiers.py 
19
20
21
22
23
24
25
26
def serialize_edge_key(edge_id: tuple[str, str, str | None]) -> str:
    """Serialize an edge id tuple to a JSON-safe string key.

    Uses a JSON array ``[source, target, relation]`` so vertex or relation names
    may contain ``|`` or other special characters without ambiguity.
    """
    source, target, relation = edge_id
    return json.dumps([source, target, relation], separators=(",", ":"))