Class GraphStore

agent/src/opentrace_agent/store/graph_store.py:148–967

Embedded graph store backed by LadybugDB. Schema:: Node(id PK, type, name, properties, search_text) RELATES(FROM Node TO Node, id STRING, type STRING, properties STRING) FTS index ``node_fts`` on ``search_text`` with Porter stemmer

class GraphStore:
    """Embedded graph store backed by LadybugDB.

    Schema::

        Node(id PK, type, name, properties, search_text)
        RELATES(FROM Node TO Node, id STRING, type STRING, properties STRING)
        FTS index ``node_fts`` on ``search_text`` with Porter stemmer
    """

    def __init__(self, db_path: str, *, read_only: bool = False) -> None:
        self._db = ladybug.Database(db_path, read_only=read_only)
        self._conn = ladybug.Connection(self._db)
        self._load_extensions()
        if not read_only:
            self._ensure_schema()

    # -- schema ----------------------------------------------------------

    def _load_extensions(self) -> None:
        """Install and load the FTS extension (required for full-text search)."""
        try:
            self._conn.execute("INSTALL FTS")
            self._conn.execute("LOAD EXTENSION FTS")
        except RuntimeError:
            pass  # already installed/loaded

    def _ensure_schema(self) -> None:
        stmts = [
            "CREATE NODE TABLE IF NOT EXISTS Node(id STRING PRIMARY KEY, type STRING, name STRING, properties STRING)",
            "CREATE REL TABLE IF NOT EXISTS RELATES(FROM Node TO Node, id STRING, type STRING, properties STRING)",
        ]
        for stmt in stmts:
            self._conn.execute(stmt)

        # Add search_text column (ALTER not idempotent — ignore if exists).
        try:
            self._conn.execute("ALTER TABLE Node ADD search_text STRING DEFAULT ''")
        except RuntimeError:
            pass  # column already exists

        # FTS index (also not idempotent).
        try:
            self._conn.execute("CALL CREATE_FTS_INDEX('Node', 'node_fts', ['search_text'], stemmer := 'porter')")
        except RuntimeError:
            pass  # index already exists

    # -- write -----------------------------------------------------------

    def add_node(
        self,
        id: str,
        node_type: str,
        name: str,
        properties: dict[str, Any] | None = None,
    ) -> None:
        """Insert or update a single node (MERGE)."""
        props_json = _marshal_props(properties)