Cómo consumir una API GraphQL en Python: paso a paso
Cada vez más plataformas — GitHub, Shopify, Contentful — exponen sus datos mediante GraphQL en lugar de REST. Consumir una API GraphQL en Python es más sencillo de lo que parece: siempre es un POST a la misma URL, con una query en texto que describe exactamente los campos que queremos. Aquí tienes el camino completo, desde la primera petición hasta un DataFrame de pandas listo para analizar.
Requisitos previos
- Python 3.9 o superior instalado.
- Las librerías
requestsypandas:pip install requests pandas. - Un endpoint GraphQL. En los ejemplos usamos el público
https://countries.trevorblades.com/graphql, que no exige autenticación. - Nociones básicas de JSON y de diccionarios en Python.
Paso 1: Entender qué cambia respecto a REST
En una API REST hay muchas URL (una por recurso) y el servidor decide qué campos devuelve. En una API GraphQL hay una única URL y un único método, el POST. Lo que cambia de petición en petición es el cuerpo: una query en la que el cliente pide solo los campos que necesita. Eso elimina el clásico problema de recibir 40 campos cuando solo importan tres.
Paso 2: Escribir la primera query
La query es solo texto. Escríbela y pruébala en el explorador del proveedor (casi todos tienen uno) antes de llevarla a Python:
query {
countries {
code
name
capital
}
}
Paso 3: Enviar la query con requests
El cuerpo de la petición es siempre un JSON con la clave query (y, opcionalmente, variables). La respuesta útil viene dentro de la clave data.
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])
Fíjate en dos detalles que evitan dolores de cabeza: json= (y no data=) se encarga de la serialización y de la cabecera correcta, y timeout impide que el script se quede colgado para siempre.
Paso 4: Usar variables en lugar de concatenar cadenas
Nunca construyas la query con f-strings. Declara variables con tipo y envíalas en el campo variables: es más seguro y el servidor puede reutilizar el plan de la query.
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))
Paso 5: Autenticarse con un token
Las API privadas piden un token en la cabecera Authorization. Guárdalo en una variable de entorno, nunca en el código:
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"])
Paso 6: Tratar el error disfrazado de éxito
Esta es la trampa número uno: una API GraphQL devuelve HTTP 200 incluso cuando la query falla. raise_for_status() no detecta nada: el error está en la clave errors del JSON. Centraliza todo en una función:
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"]
Si recibesKeyError: 'data', casi con toda seguridad la respuesta traíaerrors. Imprime el JSON completo antes de culpar a la red.
Paso 7: Paginar con cursores
GraphQL no usa ?page=1. Se pide un bloque de N registros y el servidor devuelve, dentro de pageInfo, un endCursor y un hasNextPage. El bucle se repite hasta que hasNextPage sea falso:
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))
Pedir bloques demasiado grandes (por ejemplo 1000) suele provocar un error de coste de query. Entre 50 y 100 es el rango seguro en la mayoría de las API.
Paso 8: Convertir el resultado en un DataFrame
Como la respuesta es JSON anidado, json_normalize aplana los niveles y deja la tabla lista para pandas:
import pandas as pd
df = pd.json_normalize(all_repos)
df.to_csv("repos.csv", index=False)
print(df.head())
Comprobar el resultado
Antes de dar el script por terminado, confirma tres cosas:
- El
printdel primer paso muestra registros reales (y no una lista vacía). - Una query rota a propósito (cambia
namepornombre) hace querun_querylanceRuntimeError: prueba de que realmente estás leyendo el campoerrors. - El número de filas del DataFrame coincide con
len(all_repos)y con el total que la plataforma muestra en su interfaz.
Conclusión
Con un POST, un diccionario de variables, una comprobación de errors y un bucle de cursores tienes todo lo necesario para extraer datos de prácticamente cualquier API GraphQL. El siguiente paso natural es programar este script (Azure Functions, Power Automate o un simple cron) y escribir el resultado en un Lakehouse o en una tabla SQL, en lugar de un CSV. Y queda la pregunta: de las API que consumes hoy, ¿cuántos campos estás recibiendo sin usarlos nunca?