"""Rover CLI wrapper for fetching subgraph SDL from Apollo GraphOS.""" import json import subprocess def list_subgraphs(graph_ref: str) -> list[str]: """List all subgraph names for a graph ref (e.g. 'my-graph@prod'). Uses `rover subgraph list --format json` for reliable parsing. """ result = subprocess.run( ['rover', 'subgraph', 'list', graph_ref, '--format', 'json'], capture_output=True, text=True, ) if result.returncode != 0: raise RuntimeError(f'rover subgraph list failed:\n{result.stderr.strip()}') payload = json.loads(result.stdout) subgraphs = payload['data']['subgraphs'] return sorted(s['name'] for s in subgraphs) def fetch_subgraph_sdl(graph_ref: str, subgraph: str) -> str: """Fetch the deployed SDL for a single subgraph.""" result = subprocess.run( ['rover', 'subgraph', 'fetch', graph_ref, '--name', subgraph], capture_output=True, text=True, ) if result.returncode != 0: raise RuntimeError(f'rover subgraph fetch failed for {subgraph}:\n{result.stderr.strip()}') sdl = result.stdout if not sdl.strip(): raise RuntimeError(f'rover returned empty SDL for {subgraph}') return sdl