How to Consume a GraphQL API in Python: Step by Step
More and more platforms — GitHub, Shopify, Contentful — expose their data through GraphQL instead of REST. Consuming a GraphQL API in Python is simpler than it looks: it is always a POST to the same URL, carrying a text query that describes exactly the fields you want. Here is the full path, from the first request to a pandas DataFrame ready to analyse.
Prerequisites
- Python 3.9 or later installed.
- The
requestsandpandaslibraries:pip install requests pandas. - A GraphQL endpoint. The examples use the public
https://countries.trevorblades.com/graphql, which requires no authentication. - Basic knowledge of JSON and Python dictionaries.
Step 1: Understand what changes compared to REST
In a REST API there are many URLs (one per resource) and the server decides which fields come back. In a GraphQL API there is a single URL and a single method, POST. What changes from request to request is the body: a query in which the client asks only for the fields it needs. That removes the classic problem of receiving 40 fields when only three matter.
Step 2: Write your first query
The query is just text. Write it and test it in the provider's explorer (almost all of them have one) before taking it into Python:
query {
countries {
code
name
capital
}
}
Step 3: Send the query with requests
The request body is always a JSON object with a query key (and optionally variables). The useful response lives inside the data key.
import requests
URL = "https://countries.trevorblades.com/graphql"
QUERY = """
query {
countries {
code
name
capital
}
}
"""
response = requests.post(URL, json={"query": QUERY}, timeout=30)
response.raise_for_status()
data = response.json()
print(data["data"]["countries"][:3])
Two details save you a lot of pain: json= (not data=) handles serialisation and the correct header, and timeout stops the script from hanging forever.
Step 4: Use variables instead of string concatenation
Never build the query with f-strings. Declare typed variables and send them in the variables field — it is safer, and the server can reuse the query plan.
QUERY = """
query CountriesByContinent($code: ID!) {
continent(code: $code) {
name
countries {
code
name
}
}
}
"""
variables = {"code": "EU"}
response = requests.post(
URL,
json={"query": QUERY, "variables": variables},
timeout=30,
)
countries = response.json()["data"]["continent"]["countries"]
print(len(countries))
Step 5: Authenticate with a token
Private APIs expect a token in the Authorization header. Keep it in an environment variable, never in the code:
import os
import requests
URL = "https://api.github.com/graphql"
headers = {"Authorization": f"Bearer {os.environ['GITHUB_TOKEN']}"}
QUERY = """
query {
viewer {
login
}
}
"""
response = requests.post(URL, json={"query": QUERY}, headers=headers, timeout=30)
print(response.json()["data"]["viewer"]["login"])
Step 6: Handle the error disguised as success
This is trap number one: a GraphQL API returns HTTP 200 even when the query fails. raise_for_status() detects nothing — the error sits in the JSON errors key. Centralise everything in one function:
def run_query(query, variables=None, headers=None):
response = requests.post(
URL,
json={"query": query, "variables": variables or {}},
headers=headers,
timeout=30,
)
response.raise_for_status()
body = response.json()
if "errors" in body:
messages = [e["message"] for e in body["errors"]]
raise RuntimeError("GraphQL: " + "; ".join(messages))
return body["data"]
If you getKeyError: 'data', the response almost certainly carriederrors. Print the whole JSON before blaming the network.
Step 7: Paginate with cursors
GraphQL does not use ?page=1. You ask for a block of N records and the server returns, inside pageInfo, an endCursor and a hasNextPage flag. The loop repeats until hasNextPage is false:
QUERY = """
query Repos($cursor: String) {
viewer {
repositories(first: 50, after: $cursor) {
nodes {
name
stargazerCount
}
pageInfo {
hasNextPage
endCursor
}
}
}
}
"""
all_repos = []
cursor = None
while True:
page = run_query(QUERY, {"cursor": cursor}, headers)["viewer"]["repositories"]
all_repos.extend(page["nodes"])
if not page["pageInfo"]["hasNextPage"]:
break
cursor = page["pageInfo"]["endCursor"]
print(len(all_repos))
Asking for blocks that are too large (say 1000) usually triggers a query-cost error. Between 50 and 100 is the safe range on most APIs.
Step 8: Turn the result into a DataFrame
Because the response is nested JSON, json_normalize flattens the levels and hands you a table ready for pandas:
import pandas as pd
df = pd.json_normalize(all_repos)
df.to_csv("repos.csv", index=False)
print(df.head())
Check the result
Before calling the script done, confirm three things:
- The
printin the first step shows real records (not an empty list). - A deliberately broken query (swap
namefornome) makesrun_queryraiseRuntimeError— proof that you really are reading theerrorsfield. - The DataFrame row count matches
len(all_repos)and the total the platform shows in its own interface.
Conclusion
With one POST, a dictionary of variables, an errors check and a cursor loop you have everything you need to extract data from virtually any GraphQL API. The natural next step is to schedule this script (Azure Functions, Power Automate or a plain cron job) and write the output into a Lakehouse or a SQL table instead of a CSV. And a question to leave you with: of the APIs you consume today, how many fields are you fetching and never using?