feat(client): add get_labels(), get_edge_types(), traverse() — R-SDK3

- LabelInfo, EdgeTypeInfo, PropertyDefinitionInfo, TraverseResult result types
- AsyncCoordinodeClient: get_labels(), get_edge_types(), traverse()
- CoordinodeClient: sync wrappers for all three methods
- traverse() maps direction strings ("outbound"/"inbound"/"both") to TraversalDirection enum
- Export new types from coordinode.__init__
- 15 mock-based unit tests in tests/unit/test_schema_crud.py (no Docker)

All 33 unit tests pass.

Author: polaz

coordinode/coordinode/__init__.py

@@ -22,7 +22,11 @@
     AsyncCoordinodeClient,
     CoordinodeClient,
     EdgeResult,
+    EdgeTypeInfo,
+    LabelInfo,
     NodeResult,
+    PropertyDefinitionInfo,
+    TraverseResult,
     VectorResult,
 )
 
@@ -36,4 +40,8 @@
     "NodeResult",
     "EdgeResult",
     "VectorResult",
+    "LabelInfo",
+    "EdgeTypeInfo",
+    "PropertyDefinitionInfo",
+    "TraverseResult",
 ]

coordinode/coordinode/client.py

@@ -85,6 +85,61 @@ def __repr__(self) -> str:
         return f"VectorResult(distance={self.distance:.4f}, node={self.node})"
 
 
+class PropertyDefinitionInfo:
+    """A property definition from the schema (name, type, required, unique)."""
+
+    def __init__(self, proto_def: Any) -> None:
+        self.name: str = proto_def.name
+        self.type: int = proto_def.type
+        self.required: bool = proto_def.required
+        self.unique: bool = proto_def.unique
+
+    def __repr__(self) -> str:
+        return (
+            f"PropertyDefinition(name={self.name!r}, type={self.type},"
+            f" required={self.required}, unique={self.unique})"
+        )
+
+
+class LabelInfo:
+    """A node label returned from the schema registry."""
+
+    def __init__(self, proto_label: Any) -> None:
+        self.name: str = proto_label.name
+        self.version: int = proto_label.version
+        self.properties: list[PropertyDefinitionInfo] = [
+            PropertyDefinitionInfo(p) for p in proto_label.properties
+        ]
+
+    def __repr__(self) -> str:
+        return f"LabelInfo(name={self.name!r}, properties={self.properties})"
+
+
+class EdgeTypeInfo:
+    """An edge type returned from the schema registry."""
+
+    def __init__(self, proto_edge_type: Any) -> None:
+        self.name: str = proto_edge_type.name
+        self.version: int = proto_edge_type.version
+        self.properties: list[PropertyDefinitionInfo] = [
+            PropertyDefinitionInfo(p) for p in proto_edge_type.properties
+        ]
+
+    def __repr__(self) -> str:
+        return f"EdgeTypeInfo(name={self.name!r}, properties={self.properties})"
+
+
+class TraverseResult:
+    """Result of a graph traversal: reached nodes and traversed edges."""
+
+    def __init__(self, proto_response: Any) -> None:
+        self.nodes: list[NodeResult] = [NodeResult(n) for n in proto_response.nodes]
+        self.edges: list[EdgeResult] = [EdgeResult(e) for e in proto_response.edges]
+
+    def __repr__(self) -> str:
+        return f"TraverseResult(nodes={len(self.nodes)}, edges={len(self.edges)})"
+
+
 # ── Async client ─────────────────────────────────────────────────────────────
 
 
@@ -303,6 +358,61 @@ async def get_schema_text(self) -> str:
 
         return "\n".join(lines)
 
+    async def get_labels(self) -> list[LabelInfo]:
+        """Return all node labels defined in the schema."""
+        from coordinode._proto.coordinode.v1.graph.schema_pb2 import ListLabelsRequest  # type: ignore[import]
+
+        resp = await self._schema_stub.ListLabels(ListLabelsRequest(), timeout=self._timeout)
+        return [LabelInfo(label) for label in resp.labels]
+
+    async def get_edge_types(self) -> list[EdgeTypeInfo]:
+        """Return all edge types defined in the schema."""
+        from coordinode._proto.coordinode.v1.graph.schema_pb2 import ListEdgeTypesRequest  # type: ignore[import]
+
+        resp = await self._schema_stub.ListEdgeTypes(ListEdgeTypesRequest(), timeout=self._timeout)
+        return [EdgeTypeInfo(et) for et in resp.edge_types]
+
+    async def traverse(
+        self,
+        start_node_id: int,
+        edge_type: str,
+        direction: str = "outbound",
+        max_depth: int = 1,
+    ) -> TraverseResult:
+        """Traverse the graph from *start_node_id* following *edge_type* edges.
+
+        Args:
+            start_node_id: ID of the node to start from.
+            edge_type: Edge type label to follow (e.g. ``"KNOWS"``).
+            direction: ``"outbound"`` (default), ``"inbound"``, or ``"both"``.
+            max_depth: Maximum hop count (default 1).
+
+        Returns:
+            :class:`TraverseResult` with ``nodes`` and ``edges`` lists.
+        """
+        from coordinode._proto.coordinode.v1.graph.graph_pb2 import (  # type: ignore[import]
+            TraversalDirection,
+            TraverseRequest,
+        )
+
+        _direction_map = {
+            "outbound": TraversalDirection.TRAVERSAL_DIRECTION_OUTBOUND,
+            "inbound": TraversalDirection.TRAVERSAL_DIRECTION_INBOUND,
+            "both": TraversalDirection.TRAVERSAL_DIRECTION_BOTH,
+        }
+        direction_value = _direction_map.get(
+            direction.lower(), TraversalDirection.TRAVERSAL_DIRECTION_OUTBOUND
+        )
+
+        req = TraverseRequest(
+            start_node_id=start_node_id,
+            edge_type=edge_type,
+            direction=direction_value,
+            max_depth=max_depth,
+        )
+        resp = await self._graph_stub.Traverse(req, timeout=self._timeout)
+        return TraverseResult(resp)
+
     async def health(self) -> bool:
         from coordinode._proto.coordinode.v1.health.health_pb2 import (  # type: ignore[import]
             HealthCheckRequest,
@@ -422,6 +532,24 @@ def create_edge(
     def get_schema_text(self) -> str:
         return self._run(self._async.get_schema_text())
 
+    def get_labels(self) -> list[LabelInfo]:
+        """Return all node labels defined in the schema."""
+        return self._run(self._async.get_labels())
+
+    def get_edge_types(self) -> list[EdgeTypeInfo]:
+        """Return all edge types defined in the schema."""
+        return self._run(self._async.get_edge_types())
+
+    def traverse(
+        self,
+        start_node_id: int,
+        edge_type: str,
+        direction: str = "outbound",
+        max_depth: int = 1,
+    ) -> TraverseResult:
+        """Traverse the graph from *start_node_id* following *edge_type* edges."""
+        return self._run(self._async.traverse(start_node_id, edge_type, direction, max_depth))
+
     def health(self) -> bool:
         return self._run(self._async.health())
 

tests/unit/test_schema_crud.py

@@ -0,0 +1,189 @@
+"""Unit tests for R-SDK3 additions: LabelInfo, EdgeTypeInfo, TraverseResult.
+
+All tests are mock-based — no proto stubs or running server required.
+Pattern mirrors test_types.py: fake proto objects with the same attribute
+interface that real generated messages provide.
+"""
+
+from coordinode.client import (
+    EdgeResult,
+    EdgeTypeInfo,
+    LabelInfo,
+    NodeResult,
+    PropertyDefinitionInfo,
+    TraverseResult,
+)
+
+
+# ── Fake proto stubs ─────────────────────────────────────────────────────────
+
+
+class _FakePropDef:
+    """Matches proto PropertyDefinition shape."""
+
+    def __init__(self, name: str, type_: int, required: bool = False, unique: bool = False) -> None:
+        self.name = name
+        self.type = type_
+        self.required = required
+        self.unique = unique
+
+
+class _FakeLabel:
+    """Matches proto Label shape."""
+
+    def __init__(self, name: str, version: int = 1, properties=None) -> None:
+        self.name = name
+        self.version = version
+        self.properties = properties or []
+
+
+class _FakeEdgeType:
+    """Matches proto EdgeType shape."""
+
+    def __init__(self, name: str, version: int = 1, properties=None) -> None:
+        self.name = name
+        self.version = version
+        self.properties = properties or []
+
+
+class _FakeNode:
+    """Matches proto Node shape."""
+
+    def __init__(self, node_id: int, labels=None, properties=None) -> None:
+        self.node_id = node_id
+        self.labels = labels or []
+        self.properties = properties or {}
+
+
+class _FakeEdge:
+    """Matches proto Edge shape."""
+
+    def __init__(self, edge_id: int, edge_type: str, source: int, target: int, properties=None) -> None:
+        self.edge_id = edge_id
+        self.edge_type = edge_type
+        self.source_node_id = source
+        self.target_node_id = target
+        self.properties = properties or {}
+
+
+class _FakeTraverseResponse:
+    """Matches proto TraverseResponse shape."""
+
+    def __init__(self, nodes=None, edges=None) -> None:
+        self.nodes = nodes or []
+        self.edges = edges or []
+
+
+# ── PropertyDefinitionInfo ───────────────────────────────────────────────────
+
+
+class TestPropertyDefinitionInfo:
+    def test_fields_are_mapped(self):
+        # type=3 = PROPERTY_TYPE_STRING (int value from proto enum)
+        p = PropertyDefinitionInfo(_FakePropDef("name", 3, required=True, unique=False))
+        assert p.name == "name"
+        assert p.type == 3
+        assert p.required is True
+        assert p.unique is False
+
+    def test_repr_contains_name(self):
+        p = PropertyDefinitionInfo(_FakePropDef("age", 1))
+        assert "age" in repr(p)
+
+    def test_optional_flags_default_false(self):
+        p = PropertyDefinitionInfo(_FakePropDef("x", 2))
+        assert p.required is False
+        assert p.unique is False
+
+
+# ── LabelInfo ────────────────────────────────────────────────────────────────
+
+
+class TestLabelInfo:
+    def test_empty_properties(self):
+        label = LabelInfo(_FakeLabel("Person", version=2))
+        assert label.name == "Person"
+        assert label.version == 2
+        assert label.properties == []
+
+    def test_properties_are_wrapped(self):
+        props = [_FakePropDef("name", 3), _FakePropDef("age", 1)]
+        label = LabelInfo(_FakeLabel("User", properties=props))
+        assert len(label.properties) == 2
+        assert all(isinstance(p, PropertyDefinitionInfo) for p in label.properties)
+        assert label.properties[0].name == "name"
+        assert label.properties[1].name == "age"
+
+    def test_repr_contains_name(self):
+        label = LabelInfo(_FakeLabel("Movie"))
+        assert "Movie" in repr(label)
+
+    def test_version_zero(self):
+        # Schema registry may return version=0 for newly created labels.
+        label = LabelInfo(_FakeLabel("Draft", version=0))
+        assert label.version == 0
+
+
+# ── EdgeTypeInfo ─────────────────────────────────────────────────────────────
+
+
+class TestEdgeTypeInfo:
+    def test_basic_fields(self):
+        et = EdgeTypeInfo(_FakeEdgeType("KNOWS", version=1))
+        assert et.name == "KNOWS"
+        assert et.version == 1
+        assert et.properties == []
+
+    def test_properties_are_wrapped(self):
+        props = [_FakePropDef("since", 6)]  # 6 = TIMESTAMP
+        et = EdgeTypeInfo(_FakeEdgeType("FOLLOWS", properties=props))
+        assert len(et.properties) == 1
+        assert et.properties[0].name == "since"
+
+    def test_repr_contains_name(self):
+        et = EdgeTypeInfo(_FakeEdgeType("RATED"))
+        assert "RATED" in repr(et)
+
+
+# ── TraverseResult ───────────────────────────────────────────────────────────
+
+
+class TestTraverseResult:
+    def test_empty_response(self):
+        result = TraverseResult(_FakeTraverseResponse())
+        assert result.nodes == []
+        assert result.edges == []
+
+    def test_nodes_are_wrapped_as_node_results(self):
+        nodes = [_FakeNode(1, ["Person"]), _FakeNode(2, ["Movie"])]
+        result = TraverseResult(_FakeTraverseResponse(nodes=nodes))
+        assert len(result.nodes) == 2
+        assert all(isinstance(n, NodeResult) for n in result.nodes)
+        assert result.nodes[0].id == 1
+        assert result.nodes[1].id == 2
+
+    def test_edges_are_wrapped_as_edge_results(self):
+        edges = [_FakeEdge(10, "KNOWS", source=1, target=2)]
+        result = TraverseResult(_FakeTraverseResponse(edges=edges))
+        assert len(result.edges) == 1
+        assert isinstance(result.edges[0], EdgeResult)
+        assert result.edges[0].source_id == 1
+        assert result.edges[0].target_id == 2
+        assert result.edges[0].type == "KNOWS"
+
+    def test_mixed_nodes_and_edges(self):
+        nodes = [_FakeNode(1, ["A"]), _FakeNode(2, ["B"]), _FakeNode(3, ["C"])]
+        edges = [
+            _FakeEdge(10, "REL", 1, 2),
+            _FakeEdge(11, "REL", 2, 3),
+        ]
+        result = TraverseResult(_FakeTraverseResponse(nodes=nodes, edges=edges))
+        assert len(result.nodes) == 3
+        assert len(result.edges) == 2
+
+    def test_repr_shows_counts(self):
+        nodes = [_FakeNode(1, [])]
+        result = TraverseResult(_FakeTraverseResponse(nodes=nodes))
+        r = repr(result)
+        assert "1" in r  # 1 node
+        assert "0" in r  # 0 edges