Scraping GraphQL Endpoints with Python

A growing number of sites replace scattered REST endpoints with a single GraphQL API: one URL, usually /graphql, that answers precise queries describing exactly the fields you want. This guide sits within Data Extraction Patterns and APIs and covers the whole workflow — confirming the endpoint, mapping the schema with or without introspection, building queries with variables, POSTing them with httpx, and working within the cost limits, persisted-query rules and error semantics that production GraphQL servers enforce.

When to Use This Approach

GraphQL scraping is the right tool when these signals appear:

The concrete advantages over parsing rendered pages are worth stating precisely. You select fields explicitly, so one round trip can return what four REST calls would. The response shape mirrors your query, so the parser is written once and cannot drift. And GraphQL schemas change deliberately — adding a field is routine, removing one usually goes through a deprecation cycle — so queries survive front-end redesigns that would break every CSS selector on the page.

Finding the endpoint in the first place works exactly like any other hidden call; the panel-filtering and request-capture workflow is in Reverse-Engineering Private APIs in Python. If the site turns out to expose plain JSON REST endpoints instead, that simpler path is the one to take. When you do get GraphQL data back, the nested result needs reshaping before storage — the techniques are in Parsing JSON and XML Responses and the field-level coercion in Cleaning and Validating Scraped Data.

Prerequisites

You need Python 3.10 or newer and an HTTP client that speaks JSON cleanly. httpx is used throughout.

python -m pip install "httpx[http2]>=0.27"

You do not need a dedicated GraphQL library. A query is just a string you POST as JSON, and keeping it to raw httpx makes the mechanics visible rather than hiding the request behind an abstraction. Libraries such as gql add schema validation and typed results, which pay off on a large internal project and add little when you are reading someone else's API.

Step-by-Step: From Endpoint to Complete Dataset

1. Confirm the Endpoint and Method

Every GraphQL call is a POST to one URL with a JSON body containing a query field, and optionally variables and operationName. Reproduce the smallest possible query first.

import httpx

GRAPHQL_URL = "https://api.example.com/graphql"
UA = ("Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
      "(KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36")
HEADERS = {
    "User-Agent": UA,
    "Accept": "application/json",
    "Content-Type": "application/json",
}


def run_query(query: str, variables: dict | None = None,
              operation_name: str | None = None) -> dict:
    payload: dict = {"query": query}
    if variables is not None:
        payload["variables"] = variables
    if operation_name is not None:
        payload["operationName"] = operation_name
    with httpx.Client(http2=True, timeout=20) as client:
        response = client.post(GRAPHQL_URL, json=payload, headers=HEADERS)
        response.raise_for_status()
        result = response.json()
    if result.get("errors"):
        raise RuntimeError(result["errors"])
    return result["data"]


if __name__ == "__main__":
    print(run_query("{ __typename }"))

A successful {"__typename": "Query"} confirms the endpoint is live and accepting raw query strings. Two negative results are also informative. A 400 with Must provide query string means the server wants a different body key. A PersistedQueryNotFound error means raw queries are refused entirely, which changes the approach — see step 6.

Raising on a non-empty errors array inside run_query is deliberate. GraphQL returns field-level failures with HTTP 200 and a partially populated data object, so code that reads result["data"]["products"] without checking will hit TypeError: 'NoneType' object is not subscriptable at some arbitrary later point rather than at the failure.

2. Map the Schema, With or Without Introspection

Unlike REST, GraphQL can describe itself. An introspection query returns every type, field and argument the API exposes. Many production servers disable it — Apollo Server has shipped with introspection off in production since version 3 — but when it is available it is the fastest possible way to map the schema.

INTROSPECTION_QUERY = """
query IntrospectionQuery {
  __schema {
    queryType { name }
    types {
      name
      kind
      fields(includeDeprecated: false) {
        name
        args { name type { name kind ofType { name kind } } }
        type { name kind ofType { name kind ofType { name } } }
      }
    }
  }
}
"""


def type_names() -> list[str]:
    schema = run_query(INTROSPECTION_QUERY)["__schema"]
    return sorted(t["name"] for t in schema["types"] if not t["name"].startswith("__"))


def fields_of(type_name: str) -> list[str]:
    schema = run_query(INTROSPECTION_QUERY)["__schema"]
    for entry in schema["types"]:
        if entry["name"] == type_name:
            return [f["name"] for f in entry.get("fields") or []]
    return []


if __name__ == "__main__":
    print(type_names()[:20])
    print(fields_of("Product"))

The nested ofType chains are how GraphQL encodes wrappers: a field typed [Product!]! is a NON_NULL wrapping a LIST wrapping a NON_NULL wrapping the named type, so you need three levels of ofType to reach the name. Requesting fewer levels is the usual reason a schema dump shows "name": null everywhere.

When introspection is disabled the server answers with GraphQL introspection is not allowed by Apollo Server or a similar message. Fall back to reading the query strings the site's own frontend sends: open the app's JavaScript bundle and search for query and mutation , or capture a few real POST bodies from the Network tab. Every field named there provably exists, which is all you need.

3. Build Queries With Variables

Never interpolate values into a query string. GraphQL has first-class variables: declare them in the operation signature and pass them in a separate variables object.

PRODUCT_QUERY = """
query GetProduct($id: ID!) {
  product(id: $id) {
    id
    name
    sku
    price { amount currencyCode }
    inStock
    category { name slug }
  }
}
"""


def get_product(product_id: str) -> dict | None:
    data = run_query(PRODUCT_QUERY, variables={"id": product_id},
                     operation_name="GetProduct")
    return data.get("product")


if __name__ == "__main__":
    product = get_product("SKU-1024")
    if product:
        print(product["name"], product["price"]["amount"])

Interpolation is not just a quoting hazard. A value containing a " produces Syntax Error: Expected Name, found String, a value containing a newline produces an unterminated string, and an integer passed where the schema declares ID! produces Variable "$id" got invalid value 5; Expected type ID. Variables sidestep all three because the server parses the query once and coerces the values against the declared types.

operationName becomes mandatory the moment your document contains more than one named operation: the server answers Must provide operation name if query contains multiple operations. Sending it always is free insurance, and many servers log it, which makes your traffic legible rather than anonymous.

4. Handle Authentication

Protected GraphQL APIs authenticate the same way REST ones do: a bearer token in an Authorization header, or a session cookie set at login. For cookie-based sessions, sign in once and reuse the client so cookies carry over — the pattern is detailed in Managing Cookies and Sessions.

import httpx

GRAPHQL_URL = "https://api.example.com/graphql"


class GraphQLSession:
    def __init__(self, token: str) -> None:
        self.client = httpx.Client(
            http2=True,
            timeout=20,
            headers={
                "User-Agent": ("Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) "
                               "AppleWebKit/537.36 (KHTML, like Gecko) "
                               "Chrome/124.0.0.0 Safari/537.36"),
                "Accept": "application/json",
                "Content-Type": "application/json",
                "Authorization": f"Bearer {token}",
            },
        )

    def execute(self, query: str, variables: dict | None = None) -> dict:
        payload: dict = {"query": query, "variables": variables or {}}
        response = self.client.post(GRAPHQL_URL, json=payload)
        response.raise_for_status()
        result = response.json()
        for error in result.get("errors") or []:
            code = (error.get("extensions") or {}).get("code")
            if code in {"UNAUTHENTICATED", "FORBIDDEN"}:
                raise PermissionError(error.get("message", code))
        if result.get("errors") and result.get("data") is None:
            raise RuntimeError(result["errors"])
        return result.get("data") or {}

    def close(self) -> None:
        self.client.close()

The extensions.code field is the machine-readable half of a GraphQL error and the part worth branching on; message is prose and changes between server versions. Distinguishing a total failure (data is null) from a partial one (data present, some fields null with a matching entry in errors) matters: partial results are common on APIs that resolve fields independently, and discarding them wastes a request you already paid for.

5. Keep Queries Inside the Server's Cost Budget

Production GraphQL servers defend themselves with depth limits, complexity scoring, or both. Cost is multiplicative — a first: 50 connection nested inside a first: 20 connection can resolve a thousand records — so the same selection set that runs fine at one page size fails at another.

The practical rules: request only the fields you actually store, keep page sizes modest (25–50 is usually well inside any limit), and when a query is rejected reduce first before you reduce the selection set, because the multiplier dominates. If you need both a large list and deep detail, split it into a shallow list query followed by per-record detail queries; two cheap requests beat one rejected request.

6. Deal With Persisted Queries

Automatic persisted queries send a SHA-256 hash of the query instead of the query text, saving bandwidth on every request. Servers configured with allowBatchedHttpRequests off and persisted queries enforced will refuse raw query strings entirely.

import hashlib
import json
import httpx

GRAPHQL_URL = "https://api.example.com/graphql"
UA = ("Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
      "(KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36")


def persisted_payload(query: str, variables: dict) -> dict:
    digest = hashlib.sha256(query.encode("utf-8")).hexdigest()
    return {
        "operationName": None,
        "variables": variables,
        "extensions": {"persistedQuery": {"version": 1, "sha256Hash": digest}},
    }


def run_persisted(query: str, variables: dict) -> dict:
    headers = {"User-Agent": UA, "Accept": "application/json",
               "Content-Type": "application/json"}
    with httpx.Client(http2=True, timeout=20, headers=headers) as client:
        payload = persisted_payload(query, variables)
        response = client.post(GRAPHQL_URL, json=payload)
        result = response.json()
        codes = {(e.get("extensions") or {}).get("code") for e in result.get("errors") or []}
        if "PERSISTED_QUERY_NOT_FOUND" in codes:
            # register the text once, then the hash works for subsequent calls
            payload["query"] = query
            response = client.post(GRAPHQL_URL, json=payload)
            result = response.json()
        if result.get("errors"):
            raise RuntimeError(json.dumps(result["errors"])[:400])
        return result["data"]

The hash must be of the exact query text the server registered, byte for byte — a differing indent or a trailing newline produces a different digest. When a server refuses to register new documents, you are limited to the hashes the site's own bundle contains; extract them from the JavaScript and reuse them with your own variables, which is usually enough because the frontend already queries what you want.

7. Fetch Lists and Paginate

Listing queries return connections of many records, almost always paginated with cursors rather than page numbers.

LIST_QUERY = """
query GetProducts($first: Int!, $after: String) {
  products(first: $first, after: $after) {
    edges { node { id name sku } }
    pageInfo { endCursor hasNextPage }
  }
}
"""


def first_page(page_size: int = 25) -> list[dict]:
    data = run_query(LIST_QUERY, variables={"first": page_size, "after": None},
                     operation_name="GetProducts")
    return [edge["node"] for edge in data["products"]["edges"]]

The full loop — reading pageInfo, feeding endCursor back into after, stopping on hasNextPage, and guarding against servers that lie about it — is covered in Handling GraphQL Pagination and Cursors.

8. Fetch Many Records in One Request With Aliases

GraphQL forbids two fields with the same name in one selection set, which looks like a limitation until you use aliases. An alias renames a field in the response, so the same query field can appear many times with different arguments. That turns a hundred sequential detail requests into one.

def multi_product_query(ids: list[str]) -> tuple[str, dict]:
    """Build one query that fetches every id in a single round trip."""
    variable_defs = ", ".join(f"$id{i}: ID!" for i in range(len(ids)))
    selections = "\n".join(
        f'  p{i}: product(id: $id{i}) {{ id name sku price {{ amount currencyCode }} }}'
        for i in range(len(ids))
    )
    query = f"query MultiProduct({variable_defs}) {{\n{selections}\n}}"
    variables = {f"id{i}": value for i, value in enumerate(ids)}
    return query, variables


def fetch_products(ids: list[str], chunk: int = 25) -> dict[str, dict]:
    out: dict[str, dict] = {}
    for start in range(0, len(ids), chunk):
        batch = ids[start:start + chunk]
        query, variables = multi_product_query(batch)
        data = run_query(query, variables=variables, operation_name="MultiProduct")
        for key, node in data.items():
            if node:
                out[node["id"]] = node
    return out


if __name__ == "__main__":
    print(len(fetch_products(["SKU-1", "SKU-2", "SKU-3"])))

The chunk size is the important parameter. Aliased fields are counted individually by every complexity limiter, so 25 aliases each costing 8 points lands at 200 while 200 aliases lands at 1,600 and is rejected outright. Start at 20–25 and measure. Note also that a missing record resolves to null rather than raising, so the if node filter is what keeps a bad id from producing a TypeError — and comparing the length of the result to the length of the input tells you exactly how many ids were unknown, which is useful signal in its own right.

Aliases interact badly with persisted queries: because the query text is generated per batch, its hash differs every time, so a server enforcing registered documents will reject all of them. On such endpoints, stick to the fixed operations the frontend already registered.

Performance and Scaling Considerations

GraphQL collapses many REST calls into one, which cuts request volume sharply, but each query is heavier for the server to resolve. The balance points to keep in mind:

• Reuse a single Client. Every request after the first skips the TCP and TLS handshake. On a sequential crawl of a thousand queries that is minutes of wall time.
• Parallelise across queries, not within pagination. Cursor pagination is inherently sequential. Split the work by category, region or date range and run those streams concurrently behind a semaphore, using the pattern in Asynchronous Scraping with asyncio and HTTPX.
• Do not batch aggressively. Some servers accept an array of operations in one POST. It looks like a free win and often is not: one expensive operation in the array can push the whole batch past the complexity limit, failing requests that would each have succeeded alone.
• Cache during development. Iterating on parsing logic against a live endpoint is slow and rude; store raw responses to disk on the first run and replay them.
• Keep one query per operation file. Servers frequently log and rate-limit by operationName, so a stable name per logical query makes your traffic legible and keeps any per-operation budget predictable. Generating a fresh operation name per request looks like evasion and buys nothing.
• Request id on every node. Even when you do not need it, it is what lets you de-duplicate across overlapping pages and resume a partial crawl without re-reading everything.
• Watch Retry-After on 429. Complexity-limited APIs commonly rate-limit by cost points per minute rather than requests per minute, so a burst of cheap queries and a single expensive one can both trip it.

Common Errors and Fixes

400 Bad Request on POST. The JSON body is malformed or the query has a syntax error. Print the errors array — GraphQL messages carry locations with the exact line and column.

HTTP 200 with an errors key. GraphQL reports application errors inside a successful response. Always check result.get("errors") before reading result["data"], and branch on extensions.code rather than the message text.

Cannot query field "x" on type "Y". Did you mean "z"? The field name is wrong, or it was removed in a schema change. The suggestion in the message is usually right; otherwise re-run introspection or copy a current query from the frontend.

Variable "$first" of required type "Int!" was not provided. You sent variables without that key, or sent null for a non-null type. Non-null variables must be present and non-null on every call.

Query has depth 12, maximum allowed is 10. A depth limiter rejected the shape. Flatten the query or split it into two requests.

PersistedQueryNotFound. The server only recognises registered query hashes. Send the query text once alongside the hash to register it, or reuse the hashes the site's own bundle already contains.

UNAUTHENTICATED or FORBIDDEN in errors with a 200 status. The token or session is missing, expired, or lacks scope. Confirm the Authorization header is actually on the wire — a header set on the request but not the client is a common oversight.

Field "products" argument "first" of type "Int!" is required, but it was not provided. The connection requires an explicit page size and has no default. Relay-style servers commonly reject an unbounded list request outright rather than returning everything, which is a deliberate protection, not a bug.

A field that worked yesterday now returns null with no error. The field was deprecated and then stubbed out, or it resolves only for authenticated users. Query __type(name: "Product") { fields { name isDeprecated deprecationReason } } when introspection is available — deprecated fields keep answering for a while before they stop.

httpx.ReadTimeout on one specific query. A single expensive resolver is timing out server-side. Reduce first on the innermost connection before raising your own timeout.

Frequently Asked Questions

Do I need a GraphQL client library? No. A query is a string in a JSON POST body, so raw httpx is enough and keeps the request transparent, which matters when you are debugging someone else's server. Libraries such as gql add schema validation and typed results that help on large codebases but hide the wire format you are trying to reproduce.

Why does the server return a 200 status with errors? GraphQL treats field-level failures as part of a normal response, so the HTTP layer reports success while the errors array carries the problem and data may be partially populated. Always inspect the body, and branch on extensions.code rather than on the human-readable message, which changes between server versions.

What if introspection is disabled? Read the operations the site's own bundle sends. Search the JavaScript for query and mutation , or capture a few POST bodies from the Network tab. Every field named there provably exists in the schema, which is enough to build the queries you need without a full type map.

How do I fetch more than one page of results? Use cursor pagination: request pageInfo { endCursor hasNextPage }, feed endCursor back as the after variable, and loop until hasNextPage is false. Add a maximum-page guard and stop early if a page returns zero edges, because some servers report one page too many.

Why did my query stop working after I reformatted it? If the endpoint uses persisted queries, the SHA-256 hash is computed over the exact query text. Changing whitespace changes the digest, so the server no longer recognises it and answers PersistedQueryNotFound. Keep the query string byte-identical to the registered version.

Related

• Data Extraction Patterns and APIs — the parent section and the other extraction routes
• Handling GraphQL Pagination and Cursors — the complete cursor loop
• Reverse-Engineering Private APIs — finding the endpoint and reproducing its headers
• Parsing JSON and XML Responses — reshaping the nested result
• Extracting JSON-LD and Structured Data — the fallback when no API is reachable