Skip to content

graflo.architecture.graph_types.container

Graph data container for vertices and edges.

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)

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]