Graph¶

@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})"
        )