feat(graph): add REST API — panorama and neighbor query endpoints

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
openclaw 2026-03-23 16:51:39 +00:00
parent 4226ba8707
commit 4bf8a85660
3 changed files with 107 additions and 0 deletions

View File

@ -11,6 +11,8 @@ from app.modules.project.application.services import ProjectService
from app.modules.project.interfaces.http.router import router as project_router, init_router as init_project_router
from app.modules.scanner.application.services import ScanService
from app.modules.scanner.interfaces.http.router import router as scanner_router, init_router as init_scanner_router
from app.modules.graph.application.services import GraphService
from app.modules.graph.interfaces.http.router import router as graph_router, init_router as init_graph_router
def create_app() -> FastAPI:
@ -28,9 +30,14 @@ def create_app() -> FastAPI:
scan_service = ScanService()
init_scanner_router(project_service, scan_service)
# Wire Graph module
graph_service = GraphService()
init_graph_router(project_service, scan_service, graph_service)
# Register routers
app.include_router(project_router, prefix="/api")
app.include_router(scanner_router, prefix="/api")
app.include_router(graph_router, prefix="/api")
# Health check
@app.get("/api/health")

View File

@ -0,0 +1,56 @@
"""Graph HTTP router — panorama and neighbor query endpoints."""
from __future__ import annotations
from dataclasses import asdict
from fastapi import APIRouter
from app.modules.project.application.services import ProjectService
from app.modules.scanner.application.services import ScanService
from app.modules.graph.application.services import GraphService
router = APIRouter(prefix="/projects/{project_id}/graph", tags=["graph"])
_project_service: ProjectService | None = None
_scan_service: ScanService | None = None
_graph_service: GraphService | None = None
def init_router(
project_service: ProjectService,
scan_service: ScanService,
graph_service: GraphService,
) -> None:
global _project_service, _scan_service, _graph_service
_project_service = project_service
_scan_service = scan_service
_graph_service = graph_service
def _get_or_trigger_scan(project_id: str):
"""Get cached scan or trigger a new one."""
result = _scan_service.get_latest_scan(project_id)
if result is None:
project = _project_service.get_project(project_id)
result = _scan_service.scan(project)
return result
@router.get("")
def get_graph(project_id: str):
"""Build and return the full panorama graph for a project."""
_project_service.get_project(project_id) # Ensure project exists (raises 404)
scan_result = _get_or_trigger_scan(project_id)
view = _graph_service.build_panorama(scan_result)
return asdict(view)
@router.get("/nodes/{node_id}/neighbors")
def get_neighbors(project_id: str, node_id: str):
"""Return the subgraph of neighbors for a given node."""
_project_service.get_project(project_id) # Ensure project exists (raises 404)
scan_result = _get_or_trigger_scan(project_id)
view = _graph_service.build_panorama(scan_result)
neighbors = _graph_service.get_neighbors(view, node_id)
return asdict(neighbors)

View File

@ -0,0 +1,44 @@
import pytest
DESIGN_DIR = "/workspace/arch-design-agent-skill-dashboard/design"
@pytest.fixture
def project_id(client):
r = client.post("/api/projects", json={"name": "test", "design_dir": DESIGN_DIR})
return r.json()["id"]
def test_get_graph(client, project_id):
r = client.get(f"/api/projects/{project_id}/graph")
assert r.status_code == 200
data = r.json()
assert "nodes" in data
assert "edges" in data
assert "groups" in data
assert len(data["nodes"]) > 0
assert len(data["groups"]) == 5
def test_get_graph_auto_scans(client, project_id):
"""Graph endpoint should auto-scan if no cached scan exists."""
r = client.get(f"/api/projects/{project_id}/graph")
assert r.status_code == 200
def test_get_neighbors(client, project_id):
# First trigger a scan via graph endpoint
client.get(f"/api/projects/{project_id}/graph")
r = client.get(f"/api/projects/{project_id}/graph/nodes/CAP-PROJ-REG/neighbors")
assert r.status_code == 200
data = r.json()
assert "nodes" in data
assert len(data["nodes"]) > 0
def test_get_neighbors_unknown_node(client, project_id):
client.get(f"/api/projects/{project_id}/graph")
r = client.get(f"/api/projects/{project_id}/graph/nodes/NONEXISTENT/neighbors")
assert r.status_code == 200
data = r.json()
assert len(data["nodes"]) == 0