Skip to content

Graph Type System

The graph type system enforces semantic constraints on what nodes and edges are valid in a graph. It is the domain equivalent of a database schema.

Building a GraphType

Subclass GraphType and declare node_schemas and edge_schemas as class-level dicts:

from knowledge_platform.domain.graph_type import (
    AttributeField, NodeSchema, EdgeSchema, GraphType,
)

class TaskGraphType(GraphType):
    type_name = "task"

    node_schemas = {
        "Task": NodeSchema(
            type_name="Task",
            fields=(
                AttributeField("title",    str,  required=True,  description="Task title"),
                AttributeField("done",     bool, required=False, default=False),
                AttributeField("priority", int,  required=False, default=0),
            ),
            description="A single actionable task.",
        ),
    }

    edge_schemas = {
        "BlockedBy": EdgeSchema(
            type_name="BlockedBy",
            allowed_source_types=frozenset({"Task"}),
            allowed_target_types=frozenset({"Task"}),
            description="Task A is blocked by Task B.",
        ),
    }

Validation Flow

When GraphService.add_node() is called:

  1. It looks up the graph's type_name in the TypeRegistry.
  2. It calls graph_type.validate_node(type_name, attributes).
  3. validate_node finds the NodeSchema and runs schema.validate(attributes).
  4. validate() checks required fields are present and types match.
  5. If any errors are found, ValueError is raised and the node is not added.

The same flow applies to edges via validate_edge().

AttributeField

Parameter Type Default Description
name str — Attribute key
python_type type — Expected Python type for runtime validation
required bool True Whether the key must be present
default Any None Value used when absent (only if required=False)
description str "" Human-readable documentation

API Reference

knowledge_platform.domain.graph_type.AttributeField dataclass

Declaration of a single attribute on a node or edge type.

Attributes:

Name Type Description
name str

Attribute key.
python_type type

Expected Python type (used for runtime validation).
required bool

Whether the attribute must be present.
default Any

Default value when the attribute is absent (ignored if required).
description str

Human-readable documentation string.
Source code in src/knowledge_platform/domain/graph_type.py 
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
@dataclass(frozen=True)
class AttributeField:
    """Declaration of a single attribute on a node or edge type.

    Attributes:
        name: Attribute key.
        python_type: Expected Python type (used for runtime validation).
        required: Whether the attribute must be present.
        default: Default value when the attribute is absent (ignored if *required*).
        description: Human-readable documentation string.
    """

    name: str
    python_type: type
    required: bool = True
    default: Any = None
    description: str = ""

knowledge_platform.domain.graph_type.NodeSchema dataclass

Semantic schema for a node type within a :class:GraphType.

Attributes:

Name Type Description
type_name str

Unique type identifier within the graph type.
fields tuple[AttributeField, ...]

Ordered attribute field declarations.
description str

Human-readable purpose of this node type.
Source code in src/knowledge_platform/domain/graph_type.py 
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
@dataclass(frozen=True)
class NodeSchema:
    """Semantic schema for a node type within a :class:`GraphType`.

    Attributes:
        type_name: Unique type identifier within the graph type.
        fields: Ordered attribute field declarations.
        description: Human-readable purpose of this node type.
    """

    type_name: str
    fields: tuple[AttributeField, ...] = field(default_factory=tuple)
    description: str = ""

    def validate(self, attributes: dict[str, Any]) -> list[str]:
        """Return a list of validation error messages (empty if valid).

        Args:
            attributes: Node attribute dict to validate.

        Returns:
            List of human-readable error strings.  Empty means valid.
        """
        errors: list[str] = []
        for f in self.fields:
            if f.required and f.name not in attributes:
                errors.append(f"Missing required attribute '{f.name}' on node type '{self.type_name}'")
            elif f.name in attributes and not isinstance(attributes[f.name], f.python_type):
                actual = type(attributes[f.name]).__name__
                errors.append(
                    f"Attribute '{f.name}' on '{self.type_name}' expected "
                    f"{f.python_type.__name__} got {actual}"
                )
        return errors

Functions

validate 
validate(attributes: dict[str, Any]) -> list[str]

Return a list of validation error messages (empty if valid).

Parameters:

Name Type Description Default
attributes dict[str, Any]

Node attribute dict to validate.

 required

Returns:

Type Description
list[str]

List of human-readable error strings. Empty means valid.
Source code in src/knowledge_platform/domain/graph_type.py 
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
def validate(self, attributes: dict[str, Any]) -> list[str]:
    """Return a list of validation error messages (empty if valid).

    Args:
        attributes: Node attribute dict to validate.

    Returns:
        List of human-readable error strings.  Empty means valid.
    """
    errors: list[str] = []
    for f in self.fields:
        if f.required and f.name not in attributes:
            errors.append(f"Missing required attribute '{f.name}' on node type '{self.type_name}'")
        elif f.name in attributes and not isinstance(attributes[f.name], f.python_type):
            actual = type(attributes[f.name]).__name__
            errors.append(
                f"Attribute '{f.name}' on '{self.type_name}' expected "
                f"{f.python_type.__name__} got {actual}"
            )
    return errors

knowledge_platform.domain.graph_type.EdgeSchema dataclass

Semantic schema for an edge type within a :class:GraphType.

Attributes:

Name Type Description
type_name str

Unique type identifier within the graph type.
allowed_source_types frozenset[str]

Node types permitted as edge sources. Empty means any source type is allowed.
allowed_target_types frozenset[str]

Node types permitted as edge targets. Empty means any target type is allowed.
fields tuple[AttributeField, ...]

Ordered attribute field declarations.
description str

Human-readable purpose of this edge type.
Source code in src/knowledge_platform/domain/graph_type.py 
 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
@dataclass(frozen=True)
class EdgeSchema:
    """Semantic schema for an edge type within a :class:`GraphType`.

    Attributes:
        type_name: Unique type identifier within the graph type.
        allowed_source_types: Node types permitted as edge sources.  Empty
            means any source type is allowed.
        allowed_target_types: Node types permitted as edge targets.  Empty
            means any target type is allowed.
        fields: Ordered attribute field declarations.
        description: Human-readable purpose of this edge type.
    """

    type_name: str
    allowed_source_types: frozenset[str] = field(default_factory=frozenset)
    allowed_target_types: frozenset[str] = field(default_factory=frozenset)
    fields: tuple[AttributeField, ...] = field(default_factory=tuple)
    description: str = ""

    def validate(
        self,
        source_type: str,
        target_type: str,
        attributes: dict[str, Any],
    ) -> list[str]:
        """Return validation error messages for an edge.

        Args:
            source_type: Type name of the source node.
            target_type: Type name of the target node.
            attributes: Edge attribute dict to validate.

        Returns:
            List of human-readable error strings.  Empty means valid.
        """
        errors: list[str] = []
        if self.allowed_source_types and source_type not in self.allowed_source_types:
            errors.append(
                f"Edge type '{self.type_name}' does not allow source type '{source_type}'"
            )
        if self.allowed_target_types and target_type not in self.allowed_target_types:
            errors.append(
                f"Edge type '{self.type_name}' does not allow target type '{target_type}'"
            )
        for f in self.fields:
            if f.required and f.name not in attributes:
                errors.append(f"Missing required attribute '{f.name}' on edge type '{self.type_name}'")
            elif f.name in attributes and not isinstance(attributes[f.name], f.python_type):
                actual = type(attributes[f.name]).__name__
                errors.append(
                    f"Attribute '{f.name}' on edge '{self.type_name}' expected "
                    f"{f.python_type.__name__} got {actual}"
                )
        return errors

Functions

validate 
validate(source_type: str, target_type: str, attributes: dict[str, Any]) -> list[str]

Return validation error messages for an edge.

Parameters:

Name Type Description Default
source_type str

Type name of the source node.

 required
target_type str

Type name of the target node.

 required
attributes dict[str, Any]

Edge attribute dict to validate.

 required

Returns:

Type Description
list[str]

List of human-readable error strings. Empty means valid.
Source code in src/knowledge_platform/domain/graph_type.py 
 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
def validate(
    self,
    source_type: str,
    target_type: str,
    attributes: dict[str, Any],
) -> list[str]:
    """Return validation error messages for an edge.

    Args:
        source_type: Type name of the source node.
        target_type: Type name of the target node.
        attributes: Edge attribute dict to validate.

    Returns:
        List of human-readable error strings.  Empty means valid.
    """
    errors: list[str] = []
    if self.allowed_source_types and source_type not in self.allowed_source_types:
        errors.append(
            f"Edge type '{self.type_name}' does not allow source type '{source_type}'"
        )
    if self.allowed_target_types and target_type not in self.allowed_target_types:
        errors.append(
            f"Edge type '{self.type_name}' does not allow target type '{target_type}'"
        )
    for f in self.fields:
        if f.required and f.name not in attributes:
            errors.append(f"Missing required attribute '{f.name}' on edge type '{self.type_name}'")
        elif f.name in attributes and not isinstance(attributes[f.name], f.python_type):
            actual = type(attributes[f.name]).__name__
            errors.append(
                f"Attribute '{f.name}' on edge '{self.type_name}' expected "
                f"{f.python_type.__name__} got {actual}"
            )
    return errors

knowledge_platform.domain.graph_type.GraphType

Base class for all graph types.

Subclasses declare node schemas and edge schemas that together define the semantics of a graph. The graph engine uses the registered :class:GraphType to validate mutations.

Subclasses must define:

  • :attr:type_name (class-level str)
  • :attr:node_schemas (class-level dict[str, NodeSchema])
  • :attr:edge_schemas (class-level dict[str, EdgeSchema])

Example::

class DocumentGraph(GraphType):
    type_name = "document"
    node_schemas = {
        "Section": NodeSchema("Section", fields=(...,)),
    }
    edge_schemas = {
        "ChildOf": EdgeSchema("ChildOf", ...),
    }
Source code in src/knowledge_platform/domain/graph_type.py 
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
class GraphType:
    """Base class for all graph types.

    Subclasses declare node schemas and edge schemas that together define the
    semantics of a graph.  The graph engine uses the registered
    :class:`GraphType` to validate mutations.

    Subclasses **must** define:

    - :attr:`type_name` (class-level ``str``)
    - :attr:`node_schemas` (class-level ``dict[str, NodeSchema]``)
    - :attr:`edge_schemas` (class-level ``dict[str, EdgeSchema]``)

    Example::

        class DocumentGraph(GraphType):
            type_name = "document"
            node_schemas = {
                "Section": NodeSchema("Section", fields=(...,)),
            }
            edge_schemas = {
                "ChildOf": EdgeSchema("ChildOf", ...),
            }
    """

    type_name: str = ""
    node_schemas: dict[str, NodeSchema] = {}
    edge_schemas: dict[str, EdgeSchema] = {}

    def validate_node(self, type_name: str, attributes: dict[str, Any]) -> None:
        """Validate node attributes against the registered schema.

        Args:
            type_name: Node type identifier.
            attributes: Attribute dict to validate.

        Raises:
            ValueError: On the first detected schema violation.
            KeyError: If *type_name* is not registered on this graph type.
        """
        schema = self.node_schemas.get(type_name)
        if schema is None:
            raise KeyError(
                f"Node type '{type_name}' is not registered on graph type '{self.type_name}'"
            )
        errors = schema.validate(attributes)
        if errors:
            raise ValueError("; ".join(errors))

    def validate_edge(
        self,
        type_name: str,
        source_type: str,
        target_type: str,
        attributes: dict[str, Any],
    ) -> None:
        """Validate edge attributes and endpoint types against the registered schema.

        Args:
            type_name: Edge type identifier.
            source_type: Type name of the source node.
            target_type: Type name of the target node.
            attributes: Attribute dict to validate.

        Raises:
            ValueError: On the first detected schema violation.
            KeyError: If *type_name* is not registered on this graph type.
        """
        schema = self.edge_schemas.get(type_name)
        if schema is None:
            raise KeyError(
                f"Edge type '{type_name}' is not registered on graph type '{self.type_name}'"
            )
        errors = schema.validate(source_type, target_type, attributes)
        if errors:
            raise ValueError("; ".join(errors))

Functions

validate_edge 
validate_edge(type_name: str, source_type: str, target_type: str, attributes: dict[str, Any]) -> None

Validate edge attributes and endpoint types against the registered schema.

Parameters:

Name Type Description Default
type_name str

Edge type identifier.

 required
source_type str

Type name of the source node.

 required
target_type str

Type name of the target node.

 required
attributes dict[str, Any]

Attribute dict to validate.

 required

Raises:

Type Description
ValueError

On the first detected schema violation.
KeyError

If type_name is not registered on this graph type.
Source code in src/knowledge_platform/domain/graph_type.py 
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
def validate_edge(
    self,
    type_name: str,
    source_type: str,
    target_type: str,
    attributes: dict[str, Any],
) -> None:
    """Validate edge attributes and endpoint types against the registered schema.

    Args:
        type_name: Edge type identifier.
        source_type: Type name of the source node.
        target_type: Type name of the target node.
        attributes: Attribute dict to validate.

    Raises:
        ValueError: On the first detected schema violation.
        KeyError: If *type_name* is not registered on this graph type.
    """
    schema = self.edge_schemas.get(type_name)
    if schema is None:
        raise KeyError(
            f"Edge type '{type_name}' is not registered on graph type '{self.type_name}'"
        )
    errors = schema.validate(source_type, target_type, attributes)
    if errors:
        raise ValueError("; ".join(errors))
validate_node 
validate_node(type_name: str, attributes: dict[str, Any]) -> None

Validate node attributes against the registered schema.

Parameters:

Name Type Description Default
type_name str

Node type identifier.

 required
attributes dict[str, Any]

Attribute dict to validate.

 required

Raises:

Type Description
ValueError

On the first detected schema violation.
KeyError

If type_name is not registered on this graph type.
Source code in src/knowledge_platform/domain/graph_type.py 
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
def validate_node(self, type_name: str, attributes: dict[str, Any]) -> None:
    """Validate node attributes against the registered schema.

    Args:
        type_name: Node type identifier.
        attributes: Attribute dict to validate.

    Raises:
        ValueError: On the first detected schema violation.
        KeyError: If *type_name* is not registered on this graph type.
    """
    schema = self.node_schemas.get(type_name)
    if schema is None:
        raise KeyError(
            f"Node type '{type_name}' is not registered on graph type '{self.type_name}'"
        )
    errors = schema.validate(attributes)
    if errors:
        raise ValueError("; ".join(errors))