Edge

An Edge is a directed relationship between two Node instances within the same graph. Like Node, it is an immutable frozen dataclass. Use Edge.evolve() to produce an updated copy with a higher version.

Directed Graph Semantics

Edges are always directed: they have a source_id (origin) and a target_id (destination). The meaning of direction depends on the edge type declared by the GraphType. For example, in the Outline module a ParentOf edge runs from parent to child.

ParentOf: "Chapter 1" (source) ──► "Section 1.1" (target)

Creating an Edge

Edges are created via Edge.create() and must reference nodes that already exist in the graph:

from knowledge_platform.core.edge import Edge

edge = Edge.create(
    graph_id=graph.id,
    source_id=parent_node.id,
    target_id=child_node.id,
    type_name="ParentOf",
)

Validation is done at the service layer

Edge.create() itself does not check that the nodes exist or that the type is valid. Those checks happen in GraphService.add_edge() and Graph.add_edge().

Identity

Field Type Meaning
id EdgeId Globally unique identifier (UUID v4)
graph_id GraphId The owning graph
source_id NodeId Origin node
target_id NodeId Destination node
type_name str Semantic type (e.g. "ParentOf")

API Reference

knowledge_platform.core.edge.Edge dataclass

A directed relationship between two :class:~knowledge_platform.core.node.Node instances.

Like :class:~knowledge_platform.core.node.Node, edges are immutable value-objects. Use :meth:evolve to produce updated copies.

Attributes:

Name Type Description
id EdgeId

Unique identifier for this edge.
graph_id GraphId

Owning graph identifier.
source_id NodeId

Origin node.
target_id NodeId

Destination node.
type_name str

Semantic relationship type; resolved by the active :class:~knowledge_platform.domain.graph_type.GraphType.
attributes dict[str, object]

Arbitrary key-value payload; must be JSON-serialisable.
version int

Monotonically increasing change counter (starts at 1).
created_at datetime

UTC timestamp of initial creation.
updated_at datetime

UTC timestamp of the most recent update.
Source code in src/knowledge_platform/core/edge.py 
11
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
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
@dataclass(frozen=True)
class Edge:
    """A directed relationship between two :class:`~knowledge_platform.core.node.Node` instances.

    Like :class:`~knowledge_platform.core.node.Node`, edges are immutable
    value-objects.  Use :meth:`evolve` to produce updated copies.

    Attributes:
        id: Unique identifier for this edge.
        graph_id: Owning graph identifier.
        source_id: Origin node.
        target_id: Destination node.
        type_name: Semantic relationship type; resolved by the active
            :class:`~knowledge_platform.domain.graph_type.GraphType`.
        attributes: Arbitrary key-value payload; must be JSON-serialisable.
        version: Monotonically increasing change counter (starts at 1).
        created_at: UTC timestamp of initial creation.
        updated_at: UTC timestamp of the most recent update.
    """

    id: EdgeId
    graph_id: GraphId
    source_id: NodeId
    target_id: NodeId
    type_name: str
    attributes: dict[str, object] = field(default_factory=dict)
    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))

    @classmethod
    def create(
        cls,
        graph_id: GraphId,
        source_id: NodeId,
        target_id: NodeId,
        type_name: str,
        attributes: dict[str, object] | None = None,
    ) -> "Edge":
        """Factory method that creates a new directed edge with a fresh identity.

        Args:
            graph_id: Owning graph.
            source_id: Source (origin) node.
            target_id: Target (destination) node.
            type_name: Semantic type registered on the graph's
                :class:`~knowledge_platform.domain.graph_type.GraphType`.
            attributes: Optional initial attribute payload.

        Returns:
            A new :class:`Edge` with version ``1``.
        """
        now = datetime.now(timezone.utc)
        return cls(
            id=new_edge_id(),
            graph_id=graph_id,
            source_id=source_id,
            target_id=target_id,
            type_name=type_name,
            attributes=attributes or {},
            version=1,
            created_at=now,
            updated_at=now,
        )

    def evolve(self, **attribute_updates: object) -> "Edge":
        """Return a new :class:`Edge` with merged *attribute_updates*.

        The *version* is incremented and *updated_at* is refreshed.

        Args:
            **attribute_updates: Attribute keys/values to merge.

        Returns:
            Updated immutable :class:`Edge`.
        """
        merged = {**self.attributes, **attribute_updates}
        return replace(
            self,
            attributes=merged,
            version=self.version + 1,
            updated_at=datetime.now(timezone.utc),
        )

    def __repr__(self) -> str:  # pragma: no cover
        return (
            f"Edge(id={self.id!r}, type={self.type_name!r}, "
            f"{self.source_id!r} -> {self.target_id!r}, v{self.version})"
        )

Functions

create classmethod 
create(
    graph_id: GraphId,
    source_id: NodeId,
    target_id: NodeId,
    type_name: str,
    attributes: dict[str, object] | None = None,
) -> "Edge"

Factory method that creates a new directed edge with a fresh identity.

Parameters:

Name Type Description Default
graph_id GraphId

Owning graph.

 required
source_id NodeId

Source (origin) node.

 required
target_id NodeId

Target (destination) node.

 required
type_name str

Semantic type registered on the graph's :class:~knowledge_platform.domain.graph_type.GraphType.

 required
attributes dict[str, object] | None

Optional initial attribute payload.

 None

Returns:

Type Description
'Edge'

A new :class:Edge with version 1.
Source code in src/knowledge_platform/core/edge.py 
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
@classmethod
def create(
    cls,
    graph_id: GraphId,
    source_id: NodeId,
    target_id: NodeId,
    type_name: str,
    attributes: dict[str, object] | None = None,
) -> "Edge":
    """Factory method that creates a new directed edge with a fresh identity.

    Args:
        graph_id: Owning graph.
        source_id: Source (origin) node.
        target_id: Target (destination) node.
        type_name: Semantic type registered on the graph's
            :class:`~knowledge_platform.domain.graph_type.GraphType`.
        attributes: Optional initial attribute payload.

    Returns:
        A new :class:`Edge` with version ``1``.
    """
    now = datetime.now(timezone.utc)
    return cls(
        id=new_edge_id(),
        graph_id=graph_id,
        source_id=source_id,
        target_id=target_id,
        type_name=type_name,
        attributes=attributes or {},
        version=1,
        created_at=now,
        updated_at=now,
    )
evolve 
evolve(**attribute_updates: object) -> 'Edge'

Return a new :class:Edge with merged attribute_updates.

The version is incremented and updated_at is refreshed.

Parameters:

Name Type Description Default
**attribute_updates object

Attribute keys/values to merge.

 {}

Returns:

Type Description
'Edge'

Updated immutable :class:Edge.
Source code in src/knowledge_platform/core/edge.py 
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
def evolve(self, **attribute_updates: object) -> "Edge":
    """Return a new :class:`Edge` with merged *attribute_updates*.

    The *version* is incremented and *updated_at* is refreshed.

    Args:
        **attribute_updates: Attribute keys/values to merge.

    Returns:
        Updated immutable :class:`Edge`.
    """
    merged = {**self.attributes, **attribute_updates}
    return replace(
        self,
        attributes=merged,
        version=self.version + 1,
        updated_at=datetime.now(timezone.utc),
    )