Execute a GraphQL query via the gh CLI, passing the full payload as JSON on stdin.
(query: str, variables: dict)
| 67 | |
| 68 | |
| 69 | def gh_graphql(query: str, variables: dict) -> dict: |
| 70 | """Execute a GraphQL query via the gh CLI, passing the full payload as JSON on stdin.""" |
| 71 | clean_vars = {k: v for k, v in variables.items() if v is not None} |
| 72 | payload = json.dumps({"query": query, "variables": clean_vars}) |
| 73 | result = subprocess.run( |
| 74 | ["gh", "api", "graphql", "--input", "-"], |
| 75 | input=payload, |
| 76 | capture_output=True, |
| 77 | text=True, |
| 78 | ) |
| 79 | if result.returncode != 0: |
| 80 | raise RuntimeError(f"gh api graphql failed: {result.stderr}") |
| 81 | return json.loads(result.stdout) |
| 82 | |
| 83 | |
| 84 | def transform_reactions(reaction_groups: list) -> dict: |