Consumir APIs JSON en Python: requests, httpx y serialización

Consumir APIs JSON en Python: requests, httpx y serialización

Consumir APIs REST es una de las tareas más comunes en Python. El módulo json de la biblioteca estándar maneja la serialización, y las bibliotecas requests o httpx gestionan las peticiones HTTP.

1. json: serialización básica

import json

# dict → JSON string
datos = {
    "nombre": "Ana",
    "edad":   28,
    "activo": True,
    "cursos": ["Python", "FastAPI"],
    "perfil": {"ciudad": "Madrid"},
}
json_str = json.dumps(datos, indent=2, ensure_ascii=False)
print(json_str)

# JSON string → dict
vuelta = json.loads(json_str)
print(vuelta["nombre"])   # Ana

# dict → archivo JSON
with open("datos.json", "w", encoding="utf-8") as f:
    json.dump(datos, f, indent=2, ensure_ascii=False)

# archivo JSON → dict
with open("datos.json", "r", encoding="utf-8") as f:
    cargado = json.load(f)
print(cargado["cursos"])   # ['Python', 'FastAPI']

Serializar tipos no nativos

import json
from datetime import datetime, date
from decimal import Decimal

class SerializadorPersonalizado(json.JSONEncoder):
    def default(self, obj):
        if isinstance(obj, (datetime, date)):
            return obj.isoformat()
        if isinstance(obj, Decimal):
            return float(obj)
        return super().default(obj)

datos = {
    "fecha":  datetime.now(),
    "precio": Decimal("19.99"),
}
print(json.dumps(datos, cls=SerializadorPersonalizado, indent=2))

2. requests: peticiones HTTP sincrónicas

pip install requests

import requests

# GET básico
r = requests.get("https://api.github.com/users/torvalds", timeout=10)
r.raise_for_status()   # Lanza excepción si HTTP 4xx/5xx
usuario = r.json()     # dict directamente
print(usuario["name"])   # Linus Torvalds

# GET con parámetros de query
r = requests.get(
    "https://api.github.com/search/repositories",
    params={
        "q":    "python file-converter",
        "sort": "stars",
        "per_page": 5,
    },
    headers={"Accept": "application/vnd.github+json"},
    timeout=10,
)
repos = r.json()["items"]
for repo in repos:
    print(f"{repo['full_name']}: {repo['stargazers_count']} ⭐")

3. POST, PUT, DELETE

import requests

BASE = "https://jsonplaceholder.typicode.com"

# POST — crear recurso
r = requests.post(
    f"{BASE}/posts",
    json={   # Automáticamente serializa dict y añade Content-Type: application/json
        "title":  "Mi título",
        "body":   "Contenido del post",
        "userId": 1,
    },
    timeout=10,
)
r.raise_for_status()
nuevo = r.json()
print(f"Creado con id={nuevo['id']}")

# PUT — actualizar recurso completo
r = requests.put(
    f"{BASE}/posts/1",
    json={"id": 1, "title": "Actualizado", "body": "Nuevo contenido", "userId": 1},
    timeout=10,
)

# PATCH — actualización parcial
r = requests.patch(
    f"{BASE}/posts/1",
    json={"title": "Solo actualizo el título"},
    timeout=10,
)

# DELETE
r = requests.delete(f"{BASE}/posts/1", timeout=10)
print(f"DELETE: {r.status_code}")   # 200

4. Autenticación

import requests

# API Key en header
r = requests.get(
    "https://api.ejemplo.com/datos",
    headers={"Authorization": "Bearer mi-token-secreto"},
    timeout=10,
)

# API Key en query parameter
r = requests.get(
    "https://api.ejemplo.com/datos",
    params={"api_key": "mi-clave"},
    timeout=10,
)

# Basic Auth
r = requests.get(
    "https://api.ejemplo.com/protegido",
    auth=("usuario", "contraseña"),
    timeout=10,
)

# OAuth2 (Bearer token): obtener token primero
def obtener_token(client_id, client_secret):
    r = requests.post(
        "https://auth.ejemplo.com/token",
        data={
            "grant_type":    "client_credentials",
            "client_id":     client_id,
            "client_secret": client_secret,
        },
    )
    r.raise_for_status()
    return r.json()["access_token"]

token = obtener_token("mi-id", "mi-secreto")
r = requests.get(
    "https://api.ejemplo.com/datos",
    headers={"Authorization": f"Bearer {token}"},
)

5. Session con requests: reutilizar conexiones y headers

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

# Session básica
session = requests.Session()
session.headers.update({
    "Authorization": "Bearer mi-token",
    "Accept":        "application/json",
    "User-Agent":    "MiApp/1.0",
})

# Reintentos automáticos
retry_strategy = Retry(
    total=3,
    backoff_factor=1,            # Espera: 1s, 2s, 4s entre reintentos
    status_forcelist=[429, 500, 502, 503, 504],
)
adapter = HTTPAdapter(max_retries=retry_strategy)
session.mount("https://", adapter)

r = session.get("https://api.ejemplo.com/datos", timeout=10)
r.raise_for_status()
print(r.json())

6. Paginación automática

import requests

def obtener_todos(url, params=None, max_paginas=100):
    """Recorre todas las páginas de una API paginada."""
    pagina = 1
    todos  = []

    while pagina <= max_paginas:
        r = requests.get(
            url,
            params={**(params or {}), "page": pagina, "per_page": 100},
            timeout=10,
        )
        r.raise_for_status()
        datos = r.json()

        if not datos:   # Lista vacía → fin de la paginación
            break
        todos.extend(datos)
        pagina += 1

    return todos

# Ejemplo con GitHub
repos = obtener_todos(
    "https://api.github.com/users/torvalds/repos",
    params={"type": "public"},
)
print(f"Total repos: {len(repos)}")

Paginación por cursor (estilo Stripe, Twitter/X)

import requests

def paginar_por_cursor(url, headers, campo_cursor="next_cursor"):
    cursor = None
    while True:
        params = {"limit": 100}
        if cursor:
            params["cursor"] = cursor

        r = requests.get(url, params=params, headers=headers, timeout=10)
        r.raise_for_status()
        datos = r.json()

        yield from datos.get("items", [])

        cursor = datos.get(campo_cursor)
        if not cursor:
            break

7. httpx: cliente HTTP moderno con async

pip install httpx

# Síncrono (compatible con requests)
import httpx

with httpx.Client(timeout=10, headers={"Authorization": "Bearer mi-token"}) as client:
    r = client.get("https://api.github.com/user")
    r.raise_for_status()
    print(r.json()["login"])

# Asíncrono (ideal con FastAPI/asyncio)
import asyncio
import httpx

async def obtener_usuario(login: str) -> dict:
    async with httpx.AsyncClient(timeout=10) as client:
        r = await client.get(f"https://api.github.com/users/{login}")
        r.raise_for_status()
        return r.json()

async def main():
    usuarios = ["torvalds", "gvanrossum", "kennethreitz"]
    tareas   = [obtener_usuario(u) for u in usuarios]
    resultados = await asyncio.gather(*tareas)
    for dato in resultados:
        print(f"{dato['login']}: {dato.get('name')}")

asyncio.run(main())

8. Manejo de errores HTTP

import requests
import time

def peticion_robusta(url, max_reintentos=3, espera_base=1):
    for intento in range(max_reintentos):
        try:
            r = requests.get(url, timeout=15)

            if r.status_code == 429:   # Rate limit
                espera = int(r.headers.get("Retry-After", espera_base * 2 ** intento))
                print(f"Rate limit: esperando {espera}s...")
                time.sleep(espera)
                continue

            r.raise_for_status()
            return r.json()

        except requests.exceptions.Timeout:
            print(f"Timeout (intento {intento+1}/{max_reintentos})")
        except requests.exceptions.ConnectionError:
            print(f"Error de conexión (intento {intento+1}/{max_reintentos})")
            time.sleep(espera_base * 2 ** intento)
        except requests.exceptions.HTTPError as e:
            print(f"HTTP {e.response.status_code}: {url}")
            break   # No reintentar errores 4xx (excepto 429)

    raise RuntimeError(f"Fallo tras {max_reintentos} intentos: {url}")

9. Caché de respuestas con requests-cache

pip install requests-cache

import requests_cache

# Instalar caché con TTL de 1 hora
requests_cache.install_cache("mi_cache", expire_after=3600)

import requests

# La primera petición va a la red; el resto devuelve la caché
r = requests.get("https://api.github.com/repos/python/cpython")
print(f"Desde caché: {r.from_cache}")   # False la primera vez
r2 = requests.get("https://api.github.com/repos/python/cpython")
print(f"Desde caché: {r2.from_cache}")  # True

Resumen de métodos HTTP

Conversiones relacionadas

Conversiones de documento que siguen este tema:

• PDF a DOCX
• DOCX a PDF
• MD a PDF
• HTML a PDF
• ODT a PDF
• PDF a TXT