GraphService

GraphService is the concrete implementation of IGraphService. It adds a type-validation layer on top of GraphEngine: before any node or edge is persisted, it looks up the graph's GraphType in the TypeRegistry and validates the attributes against the declared schema.

Validation Flow

sequenceDiagram
    participant UI
    participant GS as GraphService
    participant TR as TypeRegistry
    participant GT as GraphType
    participant E as GraphEngine

    UI->>GS: add_node(graph_id, "OutlineItem", {title: "Intro"})
    GS->>E: get_graph(graph_id)
    E-->>GS: graph
    GS->>TR: get(graph.type_name)
    TR-->>GS: OutlineGraphType
    GS->>GT: validate_node("OutlineItem", {title: "Intro"})
    GT-->>GS: OK (no errors)
    GS->>E: add_node(node)
    E-->>GS: (persisted)
    GS-->>UI: node

If validation fails, ValueError is raised before the engine is called and no data is written.

Constructor

from knowledge_platform.services.graph_service import GraphService
from knowledge_platform.core.engine import GraphEngine
from knowledge_platform.domain.registry import TypeRegistry

service = GraphService(engine=engine, type_registry=type_registry)

Update Node: Partial Merge

update_node() performs a partial attribute merge — only the keys you pass are changed; all other existing attributes are preserved:

# Node currently has: {"title": "Old", "content": "Some text", "position": 0}
updated = service.update_node(node_id, {"title": "New Title"})
# Result: {"title": "New Title", "content": "Some text", "position": 0}

The merged result is still validated against the NodeSchema before persisting.

Error Handling

Method	Raises	Condition
create_graph	KeyError	type_name not in TypeRegistry
add_node	KeyError	Graph not found
add_node	ValueError	Schema validation failure
update_node	KeyError	Node not found in any cached graph
update_node	ValueError	Schema validation failure after merge
add_edge	KeyError	Graph, source node, or target node not found
add_edge	ValueError	Schema validation failure

API Reference

knowledge_platform.services.graph_service.GraphService

Facade that validates type constraints then delegates to :class:GraphEngine.

Parameters:

Name	Type	Description	Default
engine	GraphEngine	The graph engine for all persistence-backed mutations.	required
type_registry	TypeRegistry	Registry used to resolve graph types for validation.	required

Source code in src/knowledge_platform/services/graph_service.py

class GraphService:
    """Facade that validates type constraints then delegates to :class:`GraphEngine`.

    Args:
        engine: The graph engine for all persistence-backed mutations.
        type_registry: Registry used to resolve graph types for validation.
    """

    def __init__(self, engine: GraphEngine, type_registry: TypeRegistry) -> None:
        self._engine = engine
        self._type_registry = type_registry

    def bind_repository(self, repository: GraphRepository) -> None:
        """Switch the service to a different persistence repository.

        The desktop app manages one active workspace at a time. Rebinding the
        repository swaps the active graph engine to that workspace's SQLite
        database.
        """
        self._engine = GraphEngine(repository)

    def create_graph(
        self,
        workspace_id: WorkspaceId,
        type_name: str,
        name: str,
    ) -> Graph:
        """Create a new validated graph.

        Args:
            workspace_id: Owning workspace.
            type_name: Must be registered in the :class:`TypeRegistry`.
            name: Display name.

        Returns:
            Newly created :class:`Graph`.

        Raises:
            KeyError: If *type_name* is not registered.
        """
        self._type_registry.get(type_name)  # raises KeyError if unknown
        graph = self._engine.create_graph(workspace_id, type_name, name)
        logger.info("service.graph.created", graph_id=graph.id, type_name=type_name)
        return graph

    def get_graph(self, graph_id: GraphId) -> Graph:
        """Retrieve a graph.

        Args:
            graph_id: Target identifier.

        Returns:
            The :class:`Graph`.

        Raises:
            KeyError: If not found.
        """
        return self._engine.get_graph(graph_id)

    def list_graphs(self, workspace_id: WorkspaceId) -> list[Graph]:
        """List graphs in a workspace.

        Args:
            workspace_id: Owning workspace.

        Returns:
            List of graphs (may be empty).
        """
        return self._engine.list_graphs(workspace_id)

    def delete_graph(self, graph_id: GraphId) -> None:
        """Delete a graph.

        Args:
            graph_id: Target identifier.
        """
        self._engine.delete_graph(graph_id)
        logger.info("service.graph.deleted", graph_id=graph_id)

    def add_node(
        self,
        graph_id: GraphId,
        type_name: str,
        attributes: dict[str, object],
    ) -> Node:
        """Validate and add a node to a graph.

        Args:
            graph_id: Owning graph.
            type_name: Node type registered on the graph's type.
            attributes: Initial attribute payload.

        Returns:
            The created :class:`Node`.

        Raises:
            ValueError: On schema validation failure.
            KeyError: If graph or node type is not found.
        """
        graph = self._engine.get_graph(graph_id)
        gtype = self._type_registry.get(graph.type_name)
        gtype.validate_node(type_name, attributes)
        node = Node.create(graph_id, type_name, attributes)
        self._engine.add_node(node)
        logger.info("service.node.added", node_id=node.id, type_name=type_name)
        return node

    def update_node(self, node_id: NodeId, attributes: dict[str, object]) -> Node:
        """Update node attributes.

        Locates the owning graph, applies a partial attribute merge, and
        validates the merged result.

        Args:
            node_id: Target node.
            attributes: Attributes to merge.

        Returns:
            Updated :class:`Node`.

        Raises:
            KeyError: If node or graph is not found.
            ValueError: On schema validation failure.
        """
        # We need the graph_id from the current node; iterate loaded graphs.
        # For robustness, we load the node from any cached graph.
        node = self._find_node(node_id)
        graph = self._engine.get_graph(node.graph_id)
        gtype = self._type_registry.get(graph.type_name)
        updated = node.evolve(**attributes)
        gtype.validate_node(updated.type_name, updated.attributes)
        self._engine.update_node(updated)
        logger.info("service.node.updated", node_id=node_id)
        return updated

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

        Args:
            graph_id: Owning graph.
            node_id: Target node.
        """
        self._engine.remove_node(graph_id, node_id)
        logger.info("service.node.removed", node_id=node_id)

    def add_edge(
        self,
        graph_id: GraphId,
        source_id: NodeId,
        target_id: NodeId,
        type_name: str,
        attributes: dict[str, object],
    ) -> Edge:
        """Validate and add an edge.

        Args:
            graph_id: Owning graph.
            source_id: Origin node.
            target_id: Destination node.
            type_name: Edge type registered on the graph's type.
            attributes: Initial attribute payload.

        Returns:
            The created :class:`Edge`.

        Raises:
            ValueError: On schema validation failure.
            KeyError: If graph or edge type is not found.
        """
        graph = self._engine.get_graph(graph_id)
        gtype = self._type_registry.get(graph.type_name)
        source = graph.get_node(source_id)
        target = graph.get_node(target_id)
        gtype.validate_edge(type_name, source.type_name, target.type_name, attributes)
        edge = Edge.create(graph_id, source_id, target_id, type_name, attributes)
        self._engine.add_edge(edge)
        logger.info("service.edge.added", edge_id=edge.id, type_name=type_name)
        return edge

    def remove_edge(self, graph_id: GraphId, edge_id: EdgeId) -> None:
        """Remove an edge.

        Args:
            graph_id: Owning graph.
            edge_id: Target edge.
        """
        self._engine.remove_edge(graph_id, edge_id)
        logger.info("service.edge.removed", edge_id=edge_id)

    # ------------------------------------------------------------------
    # Internal helpers
    # ------------------------------------------------------------------

    def _find_node(self, node_id: NodeId) -> Node:
        """Find a node across all cached graphs.

        Args:
            node_id: Target node.

        Returns:
            The :class:`Node`.

        Raises:
            KeyError: If the node is not found in any cached graph.
        """
        for graph in self._engine._cache.values():
            try:
                return graph.get_node(node_id)
            except KeyError:
                continue
        raise KeyError(f"Node {node_id!r} not found in any loaded graph")

Functions

add_edge

add_edge(graph_id: GraphId, source_id: NodeId, target_id: NodeId, type_name: str, attributes: dict[str, object]) -> Edge

Validate and add an edge.

Parameters:

Name	Type	Description	Default
graph_id	GraphId	Owning graph.	required
source_id	NodeId	Origin node.	required
target_id	NodeId	Destination node.	required
type_name	str	Edge type registered on the graph's type.	required
attributes	dict[str, object]	Initial attribute payload.	required

Returns:

Type	Description
Edge	The created :class:Edge.

Raises:

Type	Description
ValueError	On schema validation failure.
KeyError	If graph or edge type is not found.

Source code in src/knowledge_platform/services/graph_service.py

def add_edge(
    self,
    graph_id: GraphId,
    source_id: NodeId,
    target_id: NodeId,
    type_name: str,
    attributes: dict[str, object],
) -> Edge:
    """Validate and add an edge.

    Args:
        graph_id: Owning graph.
        source_id: Origin node.
        target_id: Destination node.
        type_name: Edge type registered on the graph's type.
        attributes: Initial attribute payload.

    Returns:
        The created :class:`Edge`.

    Raises:
        ValueError: On schema validation failure.
        KeyError: If graph or edge type is not found.
    """
    graph = self._engine.get_graph(graph_id)
    gtype = self._type_registry.get(graph.type_name)
    source = graph.get_node(source_id)
    target = graph.get_node(target_id)
    gtype.validate_edge(type_name, source.type_name, target.type_name, attributes)
    edge = Edge.create(graph_id, source_id, target_id, type_name, attributes)
    self._engine.add_edge(edge)
    logger.info("service.edge.added", edge_id=edge.id, type_name=type_name)
    return edge

add_node

add_node(graph_id: GraphId, type_name: str, attributes: dict[str, object]) -> Node

Validate and add a node to a graph.

Parameters:

Name	Type	Description	Default
graph_id	GraphId	Owning graph.	required
type_name	str	Node type registered on the graph's type.	required
attributes	dict[str, object]	Initial attribute payload.	required

Returns:

Type	Description
Node	The created :class:Node.

Raises:

Type	Description
ValueError	On schema validation failure.
KeyError	If graph or node type is not found.

Source code in src/knowledge_platform/services/graph_service.py

def add_node(
    self,
    graph_id: GraphId,
    type_name: str,
    attributes: dict[str, object],
) -> Node:
    """Validate and add a node to a graph.

    Args:
        graph_id: Owning graph.
        type_name: Node type registered on the graph's type.
        attributes: Initial attribute payload.

    Returns:
        The created :class:`Node`.

    Raises:
        ValueError: On schema validation failure.
        KeyError: If graph or node type is not found.
    """
    graph = self._engine.get_graph(graph_id)
    gtype = self._type_registry.get(graph.type_name)
    gtype.validate_node(type_name, attributes)
    node = Node.create(graph_id, type_name, attributes)
    self._engine.add_node(node)
    logger.info("service.node.added", node_id=node.id, type_name=type_name)
    return node

bind_repository

bind_repository(repository: GraphRepository) -> None

Switch the service to a different persistence repository.

The desktop app manages one active workspace at a time. Rebinding the repository swaps the active graph engine to that workspace's SQLite database.

Source code in src/knowledge_platform/services/graph_service.py

def bind_repository(self, repository: GraphRepository) -> None:
    """Switch the service to a different persistence repository.

    The desktop app manages one active workspace at a time. Rebinding the
    repository swaps the active graph engine to that workspace's SQLite
    database.
    """
    self._engine = GraphEngine(repository)

create_graph

create_graph(workspace_id: WorkspaceId, type_name: str, name: str) -> Graph

Create a new validated graph.

Parameters:

Name	Type	Description	Default
workspace_id	WorkspaceId	Owning workspace.	required
type_name	str	Must be registered in the :class:TypeRegistry.	required
name	str	Display name.	required

Returns:

Type	Description
Graph	Newly created :class:Graph.

Raises:

Type	Description
KeyError	If type_name is not registered.

Source code in src/knowledge_platform/services/graph_service.py

def create_graph(
    self,
    workspace_id: WorkspaceId,
    type_name: str,
    name: str,
) -> Graph:
    """Create a new validated graph.

    Args:
        workspace_id: Owning workspace.
        type_name: Must be registered in the :class:`TypeRegistry`.
        name: Display name.

    Returns:
        Newly created :class:`Graph`.

    Raises:
        KeyError: If *type_name* is not registered.
    """
    self._type_registry.get(type_name)  # raises KeyError if unknown
    graph = self._engine.create_graph(workspace_id, type_name, name)
    logger.info("service.graph.created", graph_id=graph.id, type_name=type_name)
    return graph

delete_graph

delete_graph(graph_id: GraphId) -> None

Delete a graph.

Parameters:

Name	Type	Description	Default
graph_id	GraphId	Target identifier.	required

Source code in src/knowledge_platform/services/graph_service.py

def delete_graph(self, graph_id: GraphId) -> None:
    """Delete a graph.

    Args:
        graph_id: Target identifier.
    """
    self._engine.delete_graph(graph_id)
    logger.info("service.graph.deleted", graph_id=graph_id)

get_graph

get_graph(graph_id: GraphId) -> Graph

Retrieve a graph.

Parameters:

Name	Type	Description	Default
graph_id	GraphId	Target identifier.	required

Returns:

Name	Type	Description
The	Graph	class:Graph.

Raises:

Type	Description
KeyError	If not found.

Source code in src/knowledge_platform/services/graph_service.py

def get_graph(self, graph_id: GraphId) -> Graph:
    """Retrieve a graph.

    Args:
        graph_id: Target identifier.

    Returns:
        The :class:`Graph`.

    Raises:
        KeyError: If not found.
    """
    return self._engine.get_graph(graph_id)

list_graphs

list_graphs(workspace_id: WorkspaceId) -> list[Graph]

List graphs in a workspace.

Parameters:

Name	Type	Description	Default
workspace_id	WorkspaceId	Owning workspace.	required

Returns:

Type	Description
list[Graph]	List of graphs (may be empty).

Source code in src/knowledge_platform/services/graph_service.py

def list_graphs(self, workspace_id: WorkspaceId) -> list[Graph]:
    """List graphs in a workspace.

    Args:
        workspace_id: Owning workspace.

    Returns:
        List of graphs (may be empty).
    """
    return self._engine.list_graphs(workspace_id)

remove_edge

remove_edge(graph_id: GraphId, edge_id: EdgeId) -> None

Remove an edge.

Parameters:

Name	Type	Description	Default
graph_id	GraphId	Owning graph.	required
edge_id	EdgeId	Target edge.	required

Source code in src/knowledge_platform/services/graph_service.py

def remove_edge(self, graph_id: GraphId, edge_id: EdgeId) -> None:
    """Remove an edge.

    Args:
        graph_id: Owning graph.
        edge_id: Target edge.
    """
    self._engine.remove_edge(graph_id, edge_id)
    logger.info("service.edge.removed", edge_id=edge_id)

remove_node

remove_node(graph_id: GraphId, node_id: NodeId) -> None

Remove a node and its incident edges.

Parameters:

Name	Type	Description	Default
graph_id	GraphId	Owning graph.	required
node_id	NodeId	Target node.	required

Source code in src/knowledge_platform/services/graph_service.py

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

    Args:
        graph_id: Owning graph.
        node_id: Target node.
    """
    self._engine.remove_node(graph_id, node_id)
    logger.info("service.node.removed", node_id=node_id)

update_node

update_node(node_id: NodeId, attributes: dict[str, object]) -> Node

Update node attributes.

Locates the owning graph, applies a partial attribute merge, and validates the merged result.

Parameters:

Name	Type	Description	Default
node_id	NodeId	Target node.	required
attributes	dict[str, object]	Attributes to merge.	required

Returns:

Name	Type	Description
Updated	Node	class:Node.

Raises:

Type	Description
KeyError	If node or graph is not found.
ValueError	On schema validation failure.

Source code in src/knowledge_platform/services/graph_service.py

def update_node(self, node_id: NodeId, attributes: dict[str, object]) -> Node:
    """Update node attributes.

    Locates the owning graph, applies a partial attribute merge, and
    validates the merged result.

    Args:
        node_id: Target node.
        attributes: Attributes to merge.

    Returns:
        Updated :class:`Node`.

    Raises:
        KeyError: If node or graph is not found.
        ValueError: On schema validation failure.
    """
    # We need the graph_id from the current node; iterate loaded graphs.
    # For robustness, we load the node from any cached graph.
    node = self._find_node(node_id)
    graph = self._engine.get_graph(node.graph_id)
    gtype = self._type_registry.get(graph.type_name)
    updated = node.evolve(**attributes)
    gtype.validate_node(updated.type_name, updated.attributes)
    self._engine.update_node(updated)
    logger.info("service.node.updated", node_id=node_id)
    return updated