Skip to content

Graph

Graph is the primary aggregate root of the domain. It owns collections of Node and Edge instances and exposes structural query helpers. Unlike Node and Edge, Graph is mutable — adding or removing elements changes its internal state and increments its version.

Lifecycle

stateDiagram-v2
    [*] --> Empty : Graph.create()
    Empty --> HasNodes : add_node()
    HasNodes --> HasNodes : add_node() / update_node() / remove_node()
    HasNodes --> HasEdges : add_edge()
    HasEdges --> HasEdges : add_edge() / remove_edge()
    HasEdges --> HasNodes : remove_edge()
    HasNodes --> Empty : remove_node() (last)

Creating a Graph

Prefer using GraphEngine.create_graph() or GraphService.create_graph() rather than calling Graph.create() directly — the higher-level methods also persist the record and register the graph in cache.

from knowledge_platform.core.graph import Graph
from knowledge_platform.core.identifiers import new_workspace_id

ws_id = new_workspace_id()
graph = Graph.create(ws_id, type_name="outline", name="My Document")
print(graph.node_count())  # 0
print(graph.version)       # 1

Node Operations

from knowledge_platform.core.node import Node

node = Node.create(graph.id, "OutlineItem", {"title": "Intro"})
graph.add_node(node)
print(graph.node_count())  # 1

# Retrieve a specific node
same_node = graph.get_node(node.id)

# Iterate all nodes of a specific type
for n in graph.nodes(type_name="OutlineItem"):
    print(n.attributes["title"])

# Remove cascades incident edges automatically
graph.remove_node(node.id)

Edge Operations

from knowledge_platform.core.edge import Edge

edge = Edge.create(graph.id, parent.id, child.id, "ParentOf")
graph.add_edge(edge)

# Query edges by type
for e in graph.edges(type_name="ParentOf"):
    print(f"{e.source_id} -> {e.target_id}")

# Adjacency helpers
for e in graph.outgoing_edges(parent.id, type_name="ParentOf"):
    print(e.target_id)

for e in graph.incoming_edges(child.id):
    print(e.source_id)

Error Handling

Method Raises Condition
add_node(node) ValueError Node belongs to a different graph or already exists
update_node(node) KeyError Node not found
remove_node(node_id) KeyError Node not found
get_node(node_id) KeyError Node not found
add_edge(edge) ValueError Edge belongs to a different graph or already exists
add_edge(edge) KeyError Source or target node not found
remove_edge(edge_id) KeyError Edge not found
get_edge(edge_id) KeyError Edge not found

API Reference

knowledge_platform.core.graph.Graph dataclass

An in-memory graph governed by a single graph type.

:class:Graph is the primary aggregate root of the domain. It owns collections of :class:~knowledge_platform.core.node.Node and :class:~knowledge_platform.core.edge.Edge instances and exposes structural query helpers.

Attributes:

Name Type Description
id GraphId

Unique identifier.
workspace_id WorkspaceId

Owning workspace.
type_name str

The graph type controlling semantics.
name str

Human-readable display name.
version int

Incremented on each structural change.
created_at datetime

UTC creation timestamp.
updated_at datetime

UTC last-modified timestamp.
Source code in src/knowledge_platform/core/graph.py 
 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
 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
183
184
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
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
@dataclass
class Graph:
    """An in-memory graph governed by a single graph type.

    :class:`Graph` is the primary aggregate root of the domain.  It owns
    collections of :class:`~knowledge_platform.core.node.Node` and
    :class:`~knowledge_platform.core.edge.Edge` instances and exposes
    structural query helpers.

    Attributes:
        id: Unique identifier.
        workspace_id: Owning workspace.
        type_name: The graph type controlling semantics.
        name: Human-readable display name.
        version: Incremented on each structural change.
        created_at: UTC creation timestamp.
        updated_at: UTC last-modified timestamp.
    """

    id: GraphId
    workspace_id: WorkspaceId
    type_name: str
    name: str = ""
    version: int = 1
    created_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
    updated_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))

    # Internal storage – keyed by ID for O(1) lookup.
    _nodes: dict[NodeId, Node] = field(default_factory=dict, repr=False)
    _edges: dict[EdgeId, Edge] = field(default_factory=dict, repr=False)

    @classmethod
    def create(cls, workspace_id: WorkspaceId, type_name: str, name: str = "") -> "Graph":
        """Create a new empty graph.

        Args:
            workspace_id: Owning workspace.
            type_name: Semantic graph type name.
            name: Optional display name.

        Returns:
            Empty :class:`Graph`.
        """
        now = datetime.now(timezone.utc)
        return cls(
            id=new_graph_id(),
            workspace_id=workspace_id,
            type_name=type_name,
            name=name,
            version=1,
            created_at=now,
            updated_at=now,
        )

    # ------------------------------------------------------------------
    # Node mutations
    # ------------------------------------------------------------------

    def add_node(self, node: Node) -> None:
        """Insert *node* into this graph.

        Args:
            node: Must have :attr:`~Node.graph_id` matching this graph.

        Raises:
            ValueError: If the node belongs to a different graph or already exists.
        """
        if node.graph_id != self.id:
            raise ValueError(f"Node graph_id {node.graph_id!r} != {self.id!r}")
        if node.id in self._nodes:
            raise ValueError(f"Node {node.id!r} already exists in graph {self.id!r}")
        self._nodes[node.id] = node
        self._touch()

    def update_node(self, node: Node) -> None:
        """Replace an existing node with an updated version.

        Args:
            node: Updated node (same ID, higher version).

        Raises:
            KeyError: If the node does not exist in this graph.
        """
        if node.id not in self._nodes:
            raise KeyError(f"Node {node.id!r} not found in graph {self.id!r}")
        self._nodes[node.id] = node
        self._touch()

    def remove_node(self, node_id: NodeId) -> None:
        """Delete a node and all its incident edges.

        Args:
            node_id: ID of the node to remove.

        Raises:
            KeyError: If the node does not exist.
        """
        if node_id not in self._nodes:
            raise KeyError(f"Node {node_id!r} not found in graph {self.id!r}")
        del self._nodes[node_id]
        # Cascade-delete incident edges.
        to_remove = [
            eid
            for eid, e in self._edges.items()
            if e.source_id == node_id or e.target_id == node_id
        ]
        for eid in to_remove:
            del self._edges[eid]
        self._touch()

    def get_node(self, node_id: NodeId) -> Node:
        """Retrieve a node by ID.

        Args:
            node_id: Target node identifier.

        Returns:
            The :class:`~knowledge_platform.core.node.Node`.

        Raises:
            KeyError: If not found.
        """
        return self._nodes[node_id]

    def nodes(self, type_name: str | None = None) -> Iterator[Node]:
        """Iterate over nodes, optionally filtered by *type_name*.

        Args:
            type_name: If given, yield only nodes with this type.

        Yields:
            Matching :class:`~knowledge_platform.core.node.Node` instances.
        """
        for node in self._nodes.values():
            if type_name is None or node.type_name == type_name:
                yield node

    # ------------------------------------------------------------------
    # Edge mutations
    # ------------------------------------------------------------------

    def add_edge(self, edge: Edge) -> None:
        """Insert *edge* into this graph.

        Args:
            edge: Must reference nodes within this graph.

        Raises:
            ValueError: If the edge belongs to a different graph or already exists.
            KeyError: If either endpoint node is missing.
        """
        if edge.graph_id != self.id:
            raise ValueError(f"Edge graph_id {edge.graph_id!r} != {self.id!r}")
        if edge.id in self._edges:
            raise ValueError(f"Edge {edge.id!r} already exists in graph {self.id!r}")
        if edge.source_id not in self._nodes:
            raise KeyError(f"Source node {edge.source_id!r} not found")
        if edge.target_id not in self._nodes:
            raise KeyError(f"Target node {edge.target_id!r} not found")
        self._edges[edge.id] = edge
        self._touch()

    def remove_edge(self, edge_id: EdgeId) -> None:
        """Delete an edge by ID.

        Args:
            edge_id: Target edge identifier.

        Raises:
            KeyError: If not found.
        """
        if edge_id not in self._edges:
            raise KeyError(f"Edge {edge_id!r} not found in graph {self.id!r}")
        del self._edges[edge_id]
        self._touch()

    def get_edge(self, edge_id: EdgeId) -> Edge:
        """Retrieve an edge by ID.

        Args:
            edge_id: Target edge identifier.

        Returns:
            The :class:`~knowledge_platform.core.edge.Edge`.

        Raises:
            KeyError: If not found.
        """
        return self._edges[edge_id]

    def edges(self, type_name: str | None = None) -> Iterator[Edge]:
        """Iterate over edges, optionally filtered by *type_name*.

        Args:
            type_name: If given, yield only edges with this type.

        Yields:
            Matching :class:`~knowledge_platform.core.edge.Edge` instances.
        """
        for edge in self._edges.values():
            if type_name is None or edge.type_name == type_name:
                yield edge

    def outgoing_edges(self, node_id: NodeId, type_name: str | None = None) -> Iterator[Edge]:
        """Yield edges where *node_id* is the source.

        Args:
            node_id: Source node identifier.
            type_name: Optional type filter.

        Yields:
            Matching outgoing :class:`~knowledge_platform.core.edge.Edge` instances.
        """
        for edge in self._edges.values():
            if edge.source_id == node_id:
                if type_name is None or edge.type_name == type_name:
                    yield edge

    def incoming_edges(self, node_id: NodeId, type_name: str | None = None) -> Iterator[Edge]:
        """Yield edges where *node_id* is the target.

        Args:
            node_id: Target node identifier.
            type_name: Optional type filter.

        Yields:
            Matching incoming :class:`~knowledge_platform.core.edge.Edge` instances.
        """
        for edge in self._edges.values():
            if edge.target_id == node_id:
                if type_name is None or edge.type_name == type_name:
                    yield edge

    # ------------------------------------------------------------------
    # Helpers
    # ------------------------------------------------------------------

    def node_count(self) -> int:
        """Return the number of nodes."""
        return len(self._nodes)

    def edge_count(self) -> int:
        """Return the number of edges."""
        return len(self._edges)

    def _touch(self) -> None:
        self.updated_at = datetime.now(timezone.utc)
        self.version += 1

    def __repr__(self) -> str:  # pragma: no cover
        return (
            f"Graph(id={self.id!r}, type={self.type_name!r}, "
            f"nodes={self.node_count()}, edges={self.edge_count()}, v{self.version})"
        )

Functions

add_edge 
add_edge(edge: Edge) -> None

Insert edge into this graph.

Parameters:

Name Type Description Default
edge Edge

Must reference nodes within this graph.

 required

Raises:

Type Description
ValueError

If the edge belongs to a different graph or already exists.
KeyError

If either endpoint node is missing.
Source code in src/knowledge_platform/core/graph.py 
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
def add_edge(self, edge: Edge) -> None:
    """Insert *edge* into this graph.

    Args:
        edge: Must reference nodes within this graph.

    Raises:
        ValueError: If the edge belongs to a different graph or already exists.
        KeyError: If either endpoint node is missing.
    """
    if edge.graph_id != self.id:
        raise ValueError(f"Edge graph_id {edge.graph_id!r} != {self.id!r}")
    if edge.id in self._edges:
        raise ValueError(f"Edge {edge.id!r} already exists in graph {self.id!r}")
    if edge.source_id not in self._nodes:
        raise KeyError(f"Source node {edge.source_id!r} not found")
    if edge.target_id not in self._nodes:
        raise KeyError(f"Target node {edge.target_id!r} not found")
    self._edges[edge.id] = edge
    self._touch()
add_node 
add_node(node: Node) -> None

Insert node into this graph.

Parameters:

Name Type Description Default
node Node

Must have :attr:~Node.graph_id matching this graph.

 required

Raises:

Type Description
ValueError

If the node belongs to a different graph or already exists.
Source code in src/knowledge_platform/core/graph.py 
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
def add_node(self, node: Node) -> None:
    """Insert *node* into this graph.

    Args:
        node: Must have :attr:`~Node.graph_id` matching this graph.

    Raises:
        ValueError: If the node belongs to a different graph or already exists.
    """
    if node.graph_id != self.id:
        raise ValueError(f"Node graph_id {node.graph_id!r} != {self.id!r}")
    if node.id in self._nodes:
        raise ValueError(f"Node {node.id!r} already exists in graph {self.id!r}")
    self._nodes[node.id] = node
    self._touch()
create classmethod 
create(workspace_id: WorkspaceId, type_name: str, name: str = '') -> 'Graph'

Create a new empty graph.

Parameters:

Name Type Description Default
workspace_id WorkspaceId

Owning workspace.

 required
type_name str

Semantic graph type name.

 required
name str

Optional display name.

 ''

Returns:

Name Type Description
Empty 'Graph'

class:Graph.
Source code in src/knowledge_platform/core/graph.py 
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
@classmethod
def create(cls, workspace_id: WorkspaceId, type_name: str, name: str = "") -> "Graph":
    """Create a new empty graph.

    Args:
        workspace_id: Owning workspace.
        type_name: Semantic graph type name.
        name: Optional display name.

    Returns:
        Empty :class:`Graph`.
    """
    now = datetime.now(timezone.utc)
    return cls(
        id=new_graph_id(),
        workspace_id=workspace_id,
        type_name=type_name,
        name=name,
        version=1,
        created_at=now,
        updated_at=now,
    )
edge_count 
edge_count() -> int

Return the number of edges.

Source code in src/knowledge_platform/core/graph.py 
255
256
257
def edge_count(self) -> int:
    """Return the number of edges."""
    return len(self._edges)
edges 
edges(type_name: str | None = None) -> Iterator[Edge]

Iterate over edges, optionally filtered by type_name.

Parameters:

Name Type Description Default
type_name str | None

If given, yield only edges with this type.

 None

Yields:

Name Type Description
Matching Edge

class:~knowledge_platform.core.edge.Edge instances.
Source code in src/knowledge_platform/core/graph.py 
204
205
206
207
208
209
210
211
212
213
214
215
def edges(self, type_name: str | None = None) -> Iterator[Edge]:
    """Iterate over edges, optionally filtered by *type_name*.

    Args:
        type_name: If given, yield only edges with this type.

    Yields:
        Matching :class:`~knowledge_platform.core.edge.Edge` instances.
    """
    for edge in self._edges.values():
        if type_name is None or edge.type_name == type_name:
            yield edge
get_edge 
get_edge(edge_id: EdgeId) -> Edge

Retrieve an edge by ID.

Parameters:

Name Type Description Default
edge_id EdgeId

Target edge identifier.

 required

Returns:

Name Type Description
The Edge

class:~knowledge_platform.core.edge.Edge.

Raises:

Type Description
KeyError

If not found.
Source code in src/knowledge_platform/core/graph.py 
190
191
192
193
194
195
196
197
198
199
200
201
202
def get_edge(self, edge_id: EdgeId) -> Edge:
    """Retrieve an edge by ID.

    Args:
        edge_id: Target edge identifier.

    Returns:
        The :class:`~knowledge_platform.core.edge.Edge`.

    Raises:
        KeyError: If not found.
    """
    return self._edges[edge_id]
get_node 
get_node(node_id: NodeId) -> Node

Retrieve a node by ID.

Parameters:

Name Type Description Default
node_id NodeId

Target node identifier.

 required

Returns:

Name Type Description
The Node

class:~knowledge_platform.core.node.Node.

Raises:

Type Description
KeyError

If not found.
Source code in src/knowledge_platform/core/graph.py 
124
125
126
127
128
129
130
131
132
133
134
135
136
def get_node(self, node_id: NodeId) -> Node:
    """Retrieve a node by ID.

    Args:
        node_id: Target node identifier.

    Returns:
        The :class:`~knowledge_platform.core.node.Node`.

    Raises:
        KeyError: If not found.
    """
    return self._nodes[node_id]
incoming_edges 
incoming_edges(node_id: NodeId, type_name: str | None = None) -> Iterator[Edge]

Yield edges where node_id is the target.

Parameters:

Name Type Description Default
node_id NodeId

Target node identifier.

 required
type_name str | None

Optional type filter.

 None

Yields:

Type Description
Edge

Matching incoming :class:~knowledge_platform.core.edge.Edge instances.
Source code in src/knowledge_platform/core/graph.py 
232
233
234
235
236
237
238
239
240
241
242
243
244
245
def incoming_edges(self, node_id: NodeId, type_name: str | None = None) -> Iterator[Edge]:
    """Yield edges where *node_id* is the target.

    Args:
        node_id: Target node identifier.
        type_name: Optional type filter.

    Yields:
        Matching incoming :class:`~knowledge_platform.core.edge.Edge` instances.
    """
    for edge in self._edges.values():
        if edge.target_id == node_id:
            if type_name is None or edge.type_name == type_name:
                yield edge
node_count 
node_count() -> int

Return the number of nodes.

Source code in src/knowledge_platform/core/graph.py 
251
252
253
def node_count(self) -> int:
    """Return the number of nodes."""
    return len(self._nodes)
nodes 
nodes(type_name: str | None = None) -> Iterator[Node]

Iterate over nodes, optionally filtered by type_name.

Parameters:

Name Type Description Default
type_name str | None

If given, yield only nodes with this type.

 None

Yields:

Name Type Description
Matching Node

class:~knowledge_platform.core.node.Node instances.
Source code in src/knowledge_platform/core/graph.py 
138
139
140
141
142
143
144
145
146
147
148
149
def nodes(self, type_name: str | None = None) -> Iterator[Node]:
    """Iterate over nodes, optionally filtered by *type_name*.

    Args:
        type_name: If given, yield only nodes with this type.

    Yields:
        Matching :class:`~knowledge_platform.core.node.Node` instances.
    """
    for node in self._nodes.values():
        if type_name is None or node.type_name == type_name:
            yield node
outgoing_edges 
outgoing_edges(node_id: NodeId, type_name: str | None = None) -> Iterator[Edge]

Yield edges where node_id is the source.

Parameters:

Name Type Description Default
node_id NodeId

Source node identifier.

 required
type_name str | None

Optional type filter.

 None

Yields:

Type Description
Edge

Matching outgoing :class:~knowledge_platform.core.edge.Edge instances.
Source code in src/knowledge_platform/core/graph.py 
217
218
219
220
221
222
223
224
225
226
227
228
229
230
def outgoing_edges(self, node_id: NodeId, type_name: str | None = None) -> Iterator[Edge]:
    """Yield edges where *node_id* is the source.

    Args:
        node_id: Source node identifier.
        type_name: Optional type filter.

    Yields:
        Matching outgoing :class:`~knowledge_platform.core.edge.Edge` instances.
    """
    for edge in self._edges.values():
        if edge.source_id == node_id:
            if type_name is None or edge.type_name == type_name:
                yield edge
remove_edge 
remove_edge(edge_id: EdgeId) -> None

Delete an edge by ID.

Parameters:

Name Type Description Default
edge_id EdgeId

Target edge identifier.

 required

Raises:

Type Description
KeyError

If not found.
Source code in src/knowledge_platform/core/graph.py 
176
177
178
179
180
181
182
183
184
185
186
187
188
def remove_edge(self, edge_id: EdgeId) -> None:
    """Delete an edge by ID.

    Args:
        edge_id: Target edge identifier.

    Raises:
        KeyError: If not found.
    """
    if edge_id not in self._edges:
        raise KeyError(f"Edge {edge_id!r} not found in graph {self.id!r}")
    del self._edges[edge_id]
    self._touch()
remove_node 
remove_node(node_id: NodeId) -> None

Delete a node and all its incident edges.

Parameters:

Name Type Description Default
node_id NodeId

ID of the node to remove.

 required

Raises:

Type Description
KeyError

If the node does not exist.
Source code in src/knowledge_platform/core/graph.py 
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
def remove_node(self, node_id: NodeId) -> None:
    """Delete a node and all its incident edges.

    Args:
        node_id: ID of the node to remove.

    Raises:
        KeyError: If the node does not exist.
    """
    if node_id not in self._nodes:
        raise KeyError(f"Node {node_id!r} not found in graph {self.id!r}")
    del self._nodes[node_id]
    # Cascade-delete incident edges.
    to_remove = [
        eid
        for eid, e in self._edges.items()
        if e.source_id == node_id or e.target_id == node_id
    ]
    for eid in to_remove:
        del self._edges[eid]
    self._touch()
update_node 
update_node(node: Node) -> None

Replace an existing node with an updated version.

Parameters:

Name Type Description Default
node Node

Updated node (same ID, higher version).

 required

Raises:

Type Description
KeyError

If the node does not exist in this graph.
Source code in src/knowledge_platform/core/graph.py 
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
def update_node(self, node: Node) -> None:
    """Replace an existing node with an updated version.

    Args:
        node: Updated node (same ID, higher version).

    Raises:
        KeyError: If the node does not exist in this graph.
    """
    if node.id not in self._nodes:
        raise KeyError(f"Node {node.id!r} not found in graph {self.id!r}")
    self._nodes[node.id] = node
    self._touch()