Skip to content

graflo.architecture.graph_types.container

Graph data container for vertices and edges.

Attributes

Classes

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
attached dict[VertexName, list]

Documents written onto existing vertices found by a secondary identity, never creating one, by class
attached_by dict[VertexName, str]

The secondary identity each class in attached is found by
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
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 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
        attached: Documents written onto existing vertices found by a secondary
            identity, never creating one, by class
        attached_by: The secondary identity each class in ``attached`` is
            found by
    """

    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
    )
    attached: dict[VertexName, list] = Field(default_factory=dict)
    attached_by: dict[VertexName, str] = Field(default_factory=dict)

    @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 is_empty(self) -> bool:
        """Whether no vertex, attached record or edge is held.

        ``linear`` is not consulted: it only indexes the same documents per
        input record.
        """
        return not any(
            docs
            for group in (self.vertices, self.attached, self.edges)
            for docs in group.values()
        )

    def pick_unique(self):
        """Remove duplicate entries from vertices, attached documents and edges."""
        for k, v in self.vertices.items():
            self.vertices[k] = pick_unique_dict(v)
        for k, v in self.attached.items():
            self.attached[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]],
        attached: list[dict[VertexName, list]] | None = None,
        attached_by: dict[VertexName, str] | None = None,
    ) -> GraphContainer:
        """Create a GraphContainer from a list of default dictionaries.

        Args:
            list_default_dicts: List of default dictionaries containing vertex and edge data
            attached: Per-document attached documents by class, folded in order
            attached_by: The selector each attached class is found by

        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)
        adict: defaultdict[str, list] = defaultdict(list)
        for per_doc in attached or ():
            for k, v in per_doc.items():
                adict[k].extend(v)
        return GraphContainer(
            vertices=dict(vdict.items()),
            edges=dict(edict.items()),
            linear=list_default_dicts,
            attached=dict(adict.items()),
            attached_by=dict(attached_by or {}),
        )

Attributes

attached = Field(default_factory=dict) class-attribute instance-attribute
attached_by = Field(default_factory=dict) class-attribute instance-attribute
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, attached=None, attached_by=None) 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
attached list[dict[VertexName, list]] | None

Per-document attached documents by class, folded in order

 None
attached_by dict[VertexName, str] | None

The selector each attached class is found by

 None

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 
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
@classmethod
def from_docs_list(
    cls,
    list_default_dicts: list[defaultdict[GraphEntity, list]],
    attached: list[dict[VertexName, list]] | None = None,
    attached_by: dict[VertexName, str] | None = None,
) -> GraphContainer:
    """Create a GraphContainer from a list of default dictionaries.

    Args:
        list_default_dicts: List of default dictionaries containing vertex and edge data
        attached: Per-document attached documents by class, folded in order
        attached_by: The selector each attached class is found by

    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)
    adict: defaultdict[str, list] = defaultdict(list)
    for per_doc in attached or ():
        for k, v in per_doc.items():
            adict[k].extend(v)
    return GraphContainer(
        vertices=dict(vdict.items()),
        edges=dict(edict.items()),
        linear=list_default_dicts,
        attached=dict(adict.items()),
        attached_by=dict(attached_by or {}),
    )
is_empty()

Whether no vertex, attached record or edge is held.

linear is not consulted: it only indexes the same documents per input record.

Source code in graflo/architecture/graph_types/container.py 
114
115
116
117
118
119
120
121
122
123
124
def is_empty(self) -> bool:
    """Whether no vertex, attached record or edge is held.

    ``linear`` is not consulted: it only indexes the same documents per
    input record.
    """
    return not any(
        docs
        for group in (self.vertices, self.attached, self.edges)
        for docs in group.values()
    )
items()

Get an ItemsView of the container's contents.

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

Remove duplicate entries from vertices, attached documents and edges.

Source code in graflo/architecture/graph_types/container.py 
126
127
128
129
130
131
132
133
def pick_unique(self):
    """Remove duplicate entries from vertices, attached documents and edges."""
    for k, v in self.vertices.items():
        self.vertices[k] = pick_unique_dict(v)
    for k, v in self.attached.items():
        self.attached[k] = pick_unique_dict(v)
    for k, v in self.edges.items():
        self.edges[k] = pick_unique_dict(v)

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]

Functions: