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, EdgeCastingType
  • index_config — ABCFields, Weight, Index
  • container — GraphContainer
  • location — LocationIndex, ProvenancePath
  • transform — TransformPayload, VertexRep, merge helpers
  • context — ExtractionContext, AssemblyContext, observations, ActionContext

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

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)

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)

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 
52
53
54
55
56
57
58
59
60
61
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"

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

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"

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"

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"

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

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)

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

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 
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
@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 
129
130
131
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 
133
134
135
136
137
138
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

__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"

ItemsView

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

Source code in graflo/architecture/graph_types/container.py 
64
65
66
67
68
69
70
71
72
73
74
75
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]

__iter__()

Iterate over vertices and edges in the container.

Source code in graflo/architecture/graph_types/container.py 
70
71
72
73
74
75
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]

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

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

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)

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)

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)

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

TransformPayload

Bases: ConfigBaseModel

Typed transform output shared between extraction and assembly phases.

Source code in graflo/architecture/graph_types/transform.py 
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
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)

context_doc()

Named values available for weight and relation extraction.

Source code in graflo/architecture/graph_types/transform.py 
39
40
41
def context_doc(self) -> dict[str, Any]:
    """Named values available for weight and relation extraction."""
    return dict(self.named)

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

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

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

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 
44
45
46
47
48
49
50
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 
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
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

merge_row_doc_with_transform_buffer(doc, buffer_items)

Backward-compatible alias for :func:merge_observation_with_transform_buffer.

Source code in graflo/architecture/graph_types/transform.py 
77
78
79
80
81
82
def merge_row_doc_with_transform_buffer(
    doc: dict[str, Any],
    buffer_items: list[Any],
) -> dict[str, Any]:
    """Backward-compatible alias for :func:`merge_observation_with_transform_buffer`."""
    return merge_observation_with_transform_buffer(doc, buffer_items)

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=(",", ":"))