GraphQL Client with Retry and Auth

GraphQL Client with Retry and Auth

Problem Statement

Modern APIs increasingly expose GraphQL endpoints instead of REST, but shelling out to curl for a quick query is brittle: you have to hand-craft JSON bodies, remember to inject Authorization headers, and manually retry when the API rate-limits you or the network hiccups. In CI pipelines, cron jobs, and admin scripts, this leads to fragile one-liners that silently fail or spam the console.

This script provides a small, dependable CLI client that reads a .gql query file (optionally with variables), sends it to a GraphQL endpoint, and returns clean JSON on stdout. It handles bearer-token authentication, exponential backoff with jitter for transient failures, and works with both raw requests and the gql library so you can reuse it as a module.

How It Works

The script is organized into four layers:

1. Argument parsing (argparse) collects the endpoint URL, query file, variables file, auth token, and retry knobs.
2. Transport builds a requests.Session with a pre-configured Authorization header and a Retry adapter from urllib3 for connection-level retries (DNS, connection resets).
3. Application-level retry wraps the GraphQL call in a manual loop that inspects HTTP status codes (429, 5xx) and GraphQL errors for transient conditions like throttling, applying exponential backoff with jitter.
4. Execution posts the query and variables as JSON, checks for GraphQL-level errors, and prints the data payload as pretty JSON.

Two layers of retry are deliberate: urllib3 handles low-level socket failures, while the application loop handles semantic failures (rate limits, partial service degradation) that urllib3 can't reason about.

Full Script

#!/usr/bin/env python3
"""
GraphQL Client with Retry and Auth

Send a GraphQL query from a file to an endpoint, with bearer-token auth,
exponential backoff retries, and JSON output on stdout.

Usage:
    python3 graphql_client.py --url https://api.example.com/graphql \
        --query query.gql [--variables vars.json] [--token $TOKEN]
"""

from __future__ import annotations

import argparse
import json
import logging
import os
import random
import sys
import time
from pathlib import Path
from typing import Any

import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

# ---------------------------------------------------------------------------
# Configuration (env vars with sensible defaults)
# ---------------------------------------------------------------------------
DEFAULT_TIMEOUT = float(os.environ.get("GQL_TIMEOUT", "30"))
DEFAULT_MAX_ATTEMPTS = int(os.environ.get("GQL_MAX_ATTEMPTS", "4"))
DEFAULT_BACKOFF_BASE = float(os.environ.get("GQL_BACKOFF_BASE", "1.0"))
DEFAULT_BACKOFF_CAP = float(os.environ.get("GQL_BACKOFF_CAP", "30.0"))
TOKEN_ENV_VAR = os.environ.get("GQL_TOKEN_ENV", "GQL_TOKEN")

log = logging.getLogger("graphql_client")


class GraphQLError(RuntimeError):
    """Raised when the GraphQL endpoint returns errors in the response body."""


class GraphQLClient:
    """Minimal GraphQL client with auth, retries, and backoff."""

    def __init__(
        self,
        url: str,
        token: str | None = None,
        timeout: float = DEFAULT_TIMEOUT,
        max_attempts: int = DEFAULT_MAX_ATTEMPTS,
        backoff_base: float = DEFAULT_BACKOFF_BASE,
        backoff_cap: float = DEFAULT_BACKOFF_CAP,
        extra_headers: dict[str, str] | None = None,
    ) -> None:
        self.url = url
        self.token = token
        self.timeout = timeout
        self.max_attempts = max_attempts
        self.backoff_base = backoff_base
        self.backoff_cap = backoff_cap

        # Session with connection-level retries (DNS, resets, 502/503/504).
        self.session = requests.Session()
        retry = Retry(
            total=2,
            backoff_factor=0.5,
            status_forcelist=(502, 503, 504),
            allowed_methods=frozenset(["POST"]),
            raise_on_status=False,
        )
        self.session.mount("https://", HTTPAdapter(max_retries=retry))
        self.session.mount("http://", HTTPAdapter(max_retries=retry))

        headers = {
            "Content-Type": "application/json",
            "Accept": "application/json",
            "User-Agent": "graphql-client/1.0",
        }
        if token:
            headers["Authorization"] = f"Bearer {token}"
        if extra_headers:
            headers.update(extra_headers)
        self.session.headers.update(headers)

    def _sleep_backoff(self, attempt: int, retry_after: str | None = None) -> None:
        """Exponential backoff with jitter; honor Retry-After when present."""
        if retry_after and retry_after.isdigit():
            delay = float(retry_after)
        else:
            delay = min(self.backoff_cap, self.backoff_base * (2 ** attempt))
            delay += random.uniform(0, delay * 0.25)  # jitter
        log.warning("Retrying in %.2fs (attempt %d/%d)", delay, attempt + 1, self.max_attempts)
        time.sleep(delay)

    def execute(
        self,
        query: str,
        variables: dict[str, Any] | None = None,
        operation_name: str | None = None,
    ) -> dict[str, Any]:
        """Execute a GraphQL operation and return the parsed JSON response."""
        payload: dict[str, Any] = {"query": query}
        if variables:
            payload["variables"] = variables
        if operation_name:
            payload["operationName"] = operation_name

        last_exc: Exception | None = None
        for attempt in range(self.max_attempts):
            try:
                resp = self.session.post(self.url, json=payload, timeout=self.timeout)

                # Application-level retry for rate limiting / server errors.
                if resp.status_code in (429, 500, 502, 503, 504):
                    if attempt < self.max_attempts - 1:
                        self._sleep_backoff(attempt, resp.headers.get("Retry-After"))
                        continue
                    resp.raise_for_status()

                # Non-retryable HTTP error (401, 403, 404, 400...).
                if resp.status_code >= 400:
                    raise requests.HTTPError(
                        f"HTTP {resp.status_code}: {resp.text[:500]}", response=resp
                    )

                try:
                    body = resp.json()
                except ValueError as exc:
                    raise GraphQLError(f"Non-JSON response: {resp.text[:500]}") from exc

                # GraphQL errors may be partial (data + errors) or fatal.
                if "errors" in body:
                    messages = "; ".join(
                        e.get("message", "unknown") for e in body["errors"]
                    )
                    # Throttling is often surfaced as a GraphQL error, not HTTP 429.
                    if "throttl" in messages.lower() and attempt < self.max_attempts - 1:
                        self._sleep_backoff(attempt)
                        continue
                    raise GraphQLError(messages)

                return body

            except (requests.ConnectionError, requests.Timeout) as exc:
                last_exc = exc
                if attempt < self.max_attempts - 1:
                    self._sleep_backoff(attempt)
                    continue
                raise

        # Should be unreachable, but keeps type checkers happy.
        raise RuntimeError(f"Exhausted retries: {last_exc}")


def load_text(path: Path) -> str:
    """Read a text file, raising a friendly error if missing."""
    if not path.exists():
        raise FileNotFoundError(f"File not found: {path}")
    return path.read_text(encoding="utf-8")


def load_json(path: Path) -> dict[str, Any]:
    """Read a JSON file and validate that it's an object."""
    data = json.loads(load_text(path))
    if not isinstance(data, dict):
        raise ValueError(f"Variables file must contain a JSON object: {path}")
    return data


def build_parser() -> argparse.ArgumentParser:
    p = argparse.ArgumentParser(
        description="GraphQL client with retry and bearer-token auth.",
        formatter_class=argparse.ArgumentDefaultsHelpFormatter,
    )
    p.add_argument("--url", required=True, help="GraphQL endpoint URL")
    p.add_argument("--query", required=True, type=Path, help="Path to .gql query file")
    p.add_argument("--variables", type=Path, help="Optional JSON file with variables")
    p.add_argument(
        "--token",
        default=os.environ.get(TOKEN_ENV_VAR) or os.environ.get("GQL_TOKEN"),
        help=f"Bearer token (defaults to ${TOKEN_ENV_VAR})",
    )
    p.add_argument("--operation-name", help="GraphQL operationName if query has multiple")
    p.add_argument("--header", action="append", default=[], metavar="K:V",
                   help="Extra header, repeatable (e.g. --header X-Tenant:acme)")
    p.add_argument("--timeout", type=float, default=DEFAULT_TIMEOUT)
    p.add_argument("--max-attempts", type=int, default=DEFAULT_MAX_ATTEMPTS)
    p.add_argument("--backoff-base", type=float, default=DEFAULT_BACKOFF_BASE)
    p.add_argument("--backoff-cap", type=float, default=DEFAULT_BACKOFF_CAP)
    p.add_argument("--raw", action="store_true", help="Print full response envelope, not just data")
    p.add_argument("-v", "--verbose", action="store_true", help="Enable debug logging")
    return p


def parse_headers(pairs: list[str]) -> dict[str, str]:
    headers: dict[str, str] = {}
    for item in pairs:
        if ":" not in item:
            raise ValueError(f"Header must be K:V, got {item!r}")
        k, v = item.split(":", 1)
        headers[k.strip()] = v.strip()
    return headers


def main(argv: list[str] | None = None) -> int:
    args = build_parser().parse_args(argv)
    logging.basicConfig(
        level=logging.DEBUG if args.verbose else logging.INFO,
        format="%(asctime)s %(levelname)s %(name)s: %(message)s",
    )

    try:
        query = load_text(args.query)
        variables = load_json(args.variables) if args.variables else None
        extra_headers = parse_headers(args.header)
    except (FileNotFoundError, ValueError, json.JSONDecodeError) as exc:
        log.error("Input error: %s", exc)
        return 2

    client = GraphQLClient(
        url=args.url,
        token=args.token,
        timeout=args.timeout,
        max_attempts=args.max_attempts,
        backoff_base=args.backoff_base,
        backoff_cap=args.backoff_cap,
        extra_headers=extra_headers,
    )

    try:
        result = client.execute(query, variables, args.operation_name)
    except GraphQLError as exc:
        log.error("GraphQL error: %s", exc)
        return 3
    except requests.HTTPError as exc:
        log.error("HTTP error: %s", exc)
        return 4
    except requests.RequestException as exc:
        log.error("Network error after retries: %s", exc)
        return 5

    output = result if args.raw else result.get("data", result)
    json.dump(output, sys.stdout, indent=2, sort_keys=True)
    sys.stdout.write("\n")
    return 0


if __name__ == "__main__":
    sys.exit(main())

Configuration

Keep tokens out of shell history by exporting GQL_TOKEN in your environment or sourcing a .env file before running.

Usage Examples

1. Simple query with a bearer token from the environment:

export GQL_TOKEN="ghp_xxxxxxxxxxxx"
python3 graphql_client.py --url https://api.github.com/graphql --query viewer.gql

Expected output:

{
  "viewer": {
    "login": "octocat",
    "name": "The Octocat"
  }
}

2. Query with variables passed via JSON file:

cat > vars.json <<'EOF'
{ "owner": "python", "name": "cpython" }
EOF

python3 graphql_client.py \
  --url https://api.github.com/graphql \
  --query repo.gql \
  --variables vars.json \
  --header "X-GitHub-Api-Version:2022-11-28"

Expected output (truncated):

{
  "repository": {
    "stargazerCount": 61000,
    "defaultBranchRef": { "name": "main" }
  }
}

3. Debug mode with a shorter retry budget against a flaky endpoint:

GQL_MAX_ATTEMPTS=6 GQL_BACKOFF_BASE=0.5 \
python3 graphql_client.py -v \
  --url https://staging.example.com/graphql \
  --query slow_query.gql \
  --timeout 10

Expected output includes retry logs on stderr, then JSON on stdout:

2024-05-01 12:00:01 WARNING graphql_client: Retrying in 0.61s (attempt 2/6)
2024-05-01 12:00:04 WARNING graphql_client: Retrying in 1.22s (attempt 3/6)
{ "data": { ... } }

Error Handling

• Missing query file / malformed JSON variables → caught before any network call, exit code 2.
• HTTP 401 / 403 / 404 → raised as requests.HTTPError, exit code 4. The response body is included in the log for debugging.
• HTTP 429 / 5xx → retried up to max_attempts with exponential backoff; Retry-After header is honored when present.
• GraphQL-level errors in the response → raised as GraphQLError, exit code 3. If the message contains "throttl", it is treated as retryable.
• Connection resets / timeouts → retried by both urllib3 (transport) and the application loop; after exhausting attempts, exit code 5.
• Non-JSON response (e.g., an HTML error page from a proxy) → wrapped in GraphQLError with a truncated body snippet.

The exit codes are stable so you can branch on them in shell scripts: 0 success, 2 input error, 3 GraphQL error, 4 HTTP error, 5 network error.

Extending Ideas

• Add OAuth2 client-credentials flow: swap the static token for a requests_oauthlib session that refreshes the access token on 401 before retrying.
• Stream to a file or webhook: add --output -/--output file.json and pipe results into downstream tooling (jq, Kafka, S3).
• Persist queries as named operations: replace the .gql file with a directory of named operations and a --operation flag, mirroring Apollo's persisted-query workflow.
• Add a circuit breaker: track consecutive failures per endpoint in a small state file and short-circuit calls for a cooldown window, useful when many cron jobs hit the same flaky API.