cjm-graph-plugin-system 0.0.22

Defines the standardized interface and data structures for Context Graph plugins, enabling the semantic linking, decomposition, and enrichment of multi-modal content.

cjm-graph-plugin-system

Install

pip install cjm_graph_plugin_system

Project Structure

nbs/
├── utils/ (1)
│   └── mermaid.ipynb  # Convert GraphContext objects to Mermaid.js diagram strings for visualization
├── core.ipynb              # Storage-facing DTOs for Context Graph operations. The graph DATA NOUNS (`SourceRef`, `GraphNode`, `GraphEdge`, `GraphContext`) live in `cjm-context-graph-primitives` (CR-19 data-nouns-vs-storage-verbs split), as do the typed query expressions + results (`NodeQuery`/`EdgeQuery`/`RawQuery` + their result types) — all re-exported here for compatibility. **`GraphQuery` DISSOLVED at stage 4** into the typed query expressions (pass-2 Thread 5); this library retains the plugin interface until the graph-storage-adapter dissolution completes (stage 8/9).
└── plugin_interface.ipynb  # Domain-specific plugin interface for Context Graphs

Total: 3 notebooks across 1 directory

Module Dependencies

graph LR
    core["core<br/>Core Data Structures"]
    plugin_interface["plugin_interface<br/>Graph Plugin Interface"]
    utils_mermaid["utils.mermaid<br/>Mermaid Diagram Generation"]

    plugin_interface --> core
    utils_mermaid --> core

2 cross-module dependencies detected

CLI Reference

No CLI commands found in this project.

Module Overview

Detailed documentation for each module in the project:

Mermaid Diagram Generation (mermaid.ipynb)

Convert GraphContext objects to Mermaid.js diagram strings for visualization

Import

from cjm_graph_plugin_system.utils.mermaid import (
    context_to_mermaid
)

Functions

def context_to_mermaid(
    ctx: GraphContext,  # The GraphContext to visualize
    direction: str = "TD",  # Diagram direction: "TD" (top-down) or "LR" (left-right)
    node_color_map: Optional[Dict[str, str]] = None  # Map of node labels to CSS colors
) -> str:  # Mermaid.js diagram string
    "Convert a GraphContext into a Mermaid.js diagram string."

Graph Plugin Interface (plugin_interface.ipynb)

Domain-specific plugin interface for Context Graphs

Import

from cjm_graph_plugin_system.plugin_interface import (
    GraphPlugin
)

Classes

class GraphPlugin(PluginInterface):
    "Abstract base class for all Context Graph plugins."
    
    def execute(
            self,
            action: str = "get_schema",  # Action to perform (see docstring for available actions)
            **kwargs
        ) -> Dict[str, Any]:  # JSON-serializable result
        "Execute a graph operation. This is the main entry point for RemotePluginProxy.

Dispatches to the appropriate method based on `action` parameter.
All return values are JSON-serializable dictionaries for HTTP transport."
    
    def add_nodes(
            self,
            nodes: List[GraphNode]  # Nodes to create
        ) -> List[str]:  # Created node IDs
        "Bulk create nodes."
    
    def add_edges(
            self,
            edges: List[GraphEdge]  # Edges to create
        ) -> List[str]:  # Created edge IDs
        "Bulk create edges."
    
    def get_node(
            self,
            node_id: str  # UUID of node to retrieve
        ) -> Optional[GraphNode]:  # Node or None if not found
        "Get a single node by ID."
    
    def get_edge(
            self,
            edge_id: str  # UUID of edge to retrieve
        ) -> Optional[GraphEdge]:  # Edge or None if not found
        "Get a single edge by ID."
    
    def get_context(
            self,
            node_id: str,  # Starting node UUID
            depth: int = 1,  # Traversal depth (1 = immediate neighbors)
            filter_labels: Optional[List[str]] = None  # Only include nodes with these labels
        ) -> GraphContext:  # Subgraph containing node and its neighborhood
        "Get the neighborhood of a specific node."
    
    def find_nodes_by_source(
            self,
            source_ref: SourceRef  # External resource reference
        ) -> List[GraphNode]:  # Nodes attached to this source
        "Find all nodes linked to a specific external resource."
    
    def find_nodes_by_label(
            self,
            label: str,  # Node label to search for
            limit: int = 100  # Max results
        ) -> List[GraphNode]:  # Matching nodes
        "Find nodes by label."
    
    def update_node(
            self,
            node_id: str,  # UUID of node to update
            properties: Dict[str, Any]  # Properties to merge/update
        ) -> bool:  # True if successful
        "Partial update of node properties."
    
    def update_edge(
            self,
            edge_id: str,  # UUID of edge to update
            properties: Dict[str, Any]  # Properties to merge/update
        ) -> bool:  # True if successful
        "Partial update of edge properties."
    
    def delete_nodes(
            self,
            node_ids: List[str],  # UUIDs of nodes to delete
            cascade: bool = True  # Also delete connected edges
        ) -> int:  # Number of nodes deleted
        "Delete nodes (and optionally connected edges)."
    
    def delete_edges(
            self,
            edge_ids: List[str]  # UUIDs of edges to delete
        ) -> int:  # Number of edges deleted
        "Delete edges."
    
    def get_schema(self) -> Dict[str, Any]:  # Graph schema/ontology
            """Return the current ontology/schema of the graph."""
            ...
    
        @abstractmethod
        def import_graph(
            self,
            graph_data: GraphContext,  # Data to import
            merge_strategy: str = "overwrite"  # "overwrite", "skip", or "merge"
        ) -> Dict[str, int]:  # Import statistics {nodes_created, edges_created, ...}
        "Return the current ontology/schema of the graph."
    
    def import_graph(
            self,
            graph_data: GraphContext,  # Data to import
            merge_strategy: str = "overwrite"  # "overwrite", "skip", or "merge"
        ) -> Dict[str, int]:  # Import statistics {nodes_created, edges_created, ...}
        "Bulk import a GraphContext (e.g., from backup or another plugin)."
    
    def export_graph(
            self,
            filter_query: Optional[NodeQuery] = None  # Typed node filter (stage 4; GraphQuery dissolved)
        ) -> GraphContext:  # Exported subgraph or full graph
        "Export the entire graph or a filtered subset."