GraphEngine and GraphRepository¶

class GraphRepository:
    """Abstract repository interface that persistence implementations must satisfy.

    The engine delegates all I/O through this interface, keeping the engine
    itself free of SQLAlchemy or file-system concerns.
    """

    def save_graph(self, graph: Graph) -> None:
        """Persist a graph record (insert or update).

        Args:
            graph: The graph to save.
        """
        raise NotImplementedError

    def load_graph(self, graph_id: GraphId) -> Graph | None:
        """Load a graph by ID, or return ``None``.

        Args:
            graph_id: Target identifier.

        Returns:
            Loaded :class:`Graph` or ``None``.
        """
        raise NotImplementedError

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

        Args:
            workspace_id: Owning workspace.

        Returns:
            List of :class:`Graph` instances (may be empty).
        """
        raise NotImplementedError

    def delete_graph(self, graph_id: GraphId) -> None:
        """Delete a graph and all its nodes/edges.

        Args:
            graph_id: Target identifier.
        """
        raise NotImplementedError

    def save_node(self, node: Node) -> None:
        """Persist a node (insert or update).

        Args:
            node: The node to save.
        """
        raise NotImplementedError

    def delete_node(self, node_id: NodeId) -> None:
        """Delete a node.

        Args:
            node_id: Target identifier.
        """
        raise NotImplementedError

    def save_edge(self, edge: Edge) -> None:
        """Persist an edge (insert or update).

        Args:
            edge: The edge to save.
        """
        raise NotImplementedError

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

        Args:
            edge_id: Target identifier.
        """
        raise NotImplementedError

class GraphEngine:
    """Orchestrates graph operations against an in-memory cache + repository.

    :class:`GraphEngine` is the single entry-point for all graph mutations.
    It keeps a live in-memory cache of loaded graphs and persists changes
    through the injected :class:`GraphRepository`.

    Args:
        repository: The persistence back-end.
    """

    def __init__(self, repository: GraphRepository) -> None:
        self._repo = repository
        self._cache: dict[GraphId, Graph] = {}

    # ------------------------------------------------------------------
    # Graph lifecycle
    # ------------------------------------------------------------------

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

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

        Returns:
            Newly created :class:`Graph`.
        """
        graph = Graph.create(workspace_id, type_name, name)
        self._repo.save_graph(graph)
        self._cache[graph.id] = graph
        logger.info("graph.created", graph_id=graph.id, type_name=type_name)
        return graph

    def get_graph(self, graph_id: GraphId) -> Graph:
        """Return a graph from cache or repository.

        Args:
            graph_id: Target identifier.

        Returns:
            The :class:`Graph`.

        Raises:
            KeyError: If the graph does not exist.
        """
        if graph_id in self._cache:
            return self._cache[graph_id]
        graph = self._repo.load_graph(graph_id)
        if graph is None:
            raise KeyError(f"Graph {graph_id!r} not found")
        self._cache[graph_id] = graph
        return graph

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

        Args:
            workspace_id: Owning workspace.

        Returns:
            List of :class:`Graph` instances.
        """
        return self._repo.list_graphs(workspace_id)

    def delete_graph(self, graph_id: GraphId) -> None:
        """Delete a graph and remove it from cache.

        Args:
            graph_id: Target identifier.
        """
        self._cache.pop(graph_id, None)
        self._repo.delete_graph(graph_id)
        logger.info("graph.deleted", graph_id=graph_id)

    # ------------------------------------------------------------------
    # Transaction commit
    # ------------------------------------------------------------------

    def commit(self, tx: Transaction) -> None:
        """Apply all changes recorded in *tx* to the graph and repository.

        Changes are applied in recording order.  On any error the in-memory
        graph is left in a partially-applied state; callers should discard
        and reload the graph if recovery is required.

        Args:
            tx: A completed (not yet committed) :class:`Transaction`.

        Raises:
            RuntimeError: If the transaction is already committed or rolled back.
            KeyError: If the target graph is not found.
        """
        if tx.committed:
            raise RuntimeError("Transaction already committed.")
        if tx.rolled_back:
            raise RuntimeError("Cannot commit a rolled-back transaction.")

        graph = self.get_graph(tx.graph_id)
        log = logger.bind(graph_id=tx.graph_id, change_count=len(tx.changes))
        log.info("transaction.commit.start")

        for record in tx.changes:
            op = record.operation
            payload = record.payload
            if op == "add_node":
                graph.add_node(payload)
                self._repo.save_node(payload)
            elif op == "update_node":
                graph.update_node(payload)
                self._repo.save_node(payload)
            elif op == "remove_node":
                graph.remove_node(payload)
                self._repo.delete_node(payload)
            elif op == "add_edge":
                graph.add_edge(payload)
                self._repo.save_edge(payload)
            elif op == "remove_edge":
                graph.remove_edge(payload)
                self._repo.delete_edge(payload)
            else:
                raise ValueError(f"Unknown operation: {op!r}")

        self._repo.save_graph(graph)
        tx.committed = True
        log.info("transaction.commit.done")

    # ------------------------------------------------------------------
    # Convenience single-operation helpers (auto-commit)
    # ------------------------------------------------------------------

    def add_node(self, node: Node) -> None:
        """Add a node to its owning graph without an explicit transaction.

        Args:
            node: Node to add; its :attr:`~Node.graph_id` must match a loaded graph.
        """
        graph = self.get_graph(node.graph_id)
        graph.add_node(node)
        self._repo.save_node(node)
        self._repo.save_graph(graph)
        logger.debug("node.added", node_id=node.id, graph_id=node.graph_id)

    def update_node(self, node: Node) -> None:
        """Update a node in its owning graph without an explicit transaction.

        Args:
            node: Updated node.
        """
        graph = self.get_graph(node.graph_id)
        graph.update_node(node)
        self._repo.save_node(node)
        self._repo.save_graph(graph)
        logger.debug("node.updated", node_id=node.id, version=node.version)

    def remove_node(self, graph_id: GraphId, node_id: NodeId) -> None:
        """Remove a node from a graph without an explicit transaction.

        Args:
            graph_id: Owning graph.
            node_id: Node to remove.
        """
        graph = self.get_graph(graph_id)
        graph.remove_node(node_id)
        self._repo.delete_node(node_id)
        self._repo.save_graph(graph)
        logger.debug("node.removed", node_id=node_id, graph_id=graph_id)

    def add_edge(self, edge: Edge) -> None:
        """Add an edge to its owning graph without an explicit transaction.

        Args:
            edge: Edge to add.
        """
        graph = self.get_graph(edge.graph_id)
        graph.add_edge(edge)
        self._repo.save_edge(edge)
        self._repo.save_graph(graph)
        logger.debug("edge.added", edge_id=edge.id, graph_id=edge.graph_id)

    def remove_edge(self, graph_id: GraphId, edge_id: EdgeId) -> None:
        """Remove an edge from a graph without an explicit transaction.

        Args:
            graph_id: Owning graph.
            edge_id: Edge to remove.
        """
        graph = self.get_graph(graph_id)
        graph.remove_edge(edge_id)
        self._repo.delete_edge(edge_id)
        self._repo.save_graph(graph)
        logger.debug("edge.removed", edge_id=edge_id, graph_id=graph_id)