Data Acquisition II (APIs)

Si no puedes repetir la extracción mañana y obtener el mismo dataset, no tienes un dataset: tienes una descarga.

— Principio de adquisición vía API

4.1 Fundamentos de APIs para la Ciencia de Datos

Una API (Application Programming Interface) es un contrato: una dirección HTTP que responde con datos si la llamas con los parámetros y credenciales correctos. La mayoría de las APIs de datos son REST: cada recurso tiene una URL y el método HTTP indica la intención.

https://pokeapi.co/api/v2/pokemon?limit=20&offset=0
└── base ─────────┘└── recurso ─┘└── query params ───┘

Los 5 conceptos que siempre aparecen

Concepto	Qué es	Ejemplo
Recurso / endpoint	URL que representa una colección o un ítem	/pokemon, /pokemon/pikachu
Método HTTP	Intención de la operación	GET leer, POST crear
Query params	Filtros en la URL después de ?	?limit=20&offset=0
Headers	Metadatos de la petición (auth, formato)	Authorization: Bearer TOKEN
Status code	Resultado de la operación	200 ok, 404 no existe, 429 límite excedido

En Data Science casi siempre usas GET: pedir datos sin modificar nada en el servidor.

Códigos de estado que debes memorizar

Código	Significado	Qué hacer en tu script
200	Éxito	Procesar el JSON
401	Sin autenticación válida	Revisar token, no reintentar a ciegas
404	Recurso inexistente	Registrar y continuar con el siguiente ítem
429	Rate limit excedido	Esperar (Retry-After) y reintentar con backoff
500, 502, 503	Error del servidor	Reintentar con espera; si persiste, abortar con mensaje claro

Paginación: por qué una sola llamada nunca trae todo

Ninguna API seria devuelve 100.000 registros de una vez. Los divide en páginas. Hay tres patrones:

1. Offset/limit → ?limit=100&offset=200   (típico en APIs abiertas)
2. Page/per_page → ?page=3&per_page=50    (típico en GitHub, JSONPlaceholder)
3. Cursor/next URL → {"next": "https://...offset=40"}  (PokeAPI, muchas APIs modernas)

Si solo lees la primera página, tu dataset está incompleto aunque el código no falle. Siempre define el criterio de corte: página vacía, next == null, o número máximo de páginas.

4.2 Peticiones HTTP con Requests en Python

requests es la librería estándar de facto. El patrón mínimo tiene 4 pasos: pedir, verificar, parsear, usar.

import requests

# 1. Pedir con timeout SIEMPRE (sin timeout el script puede colgarse para siempre)
respuesta = requests.get("https://pokeapi.co/api/v2/pokemon/ditto", timeout=10)

# 2. Verificar: lanza excepción si el status es 4xx/5xx
respuesta.raise_for_status()

# 3. Parsear JSON a dict/list de Python
datos = respuesta.json()

# 4. Usar
print(datos["name"])            # ditto
print(datos["base_experience"]) # 101

GET con query params (la forma correcta)

import requests

# requests construye la URL por ti y escapa caracteres especiales
params = {"limit": 5, "offset": 0}
r = requests.get("https://pokeapi.co/api/v2/pokemon", params=params, timeout=10)
r.raise_for_status()

data = r.json()
print(data["count"])  # total disponible en el servidor, ej: 1302
for item in data["results"]:
    print(item["name"], "→", item["url"])

Salida esperada:

1302
bulbasaur → https://pokeapi.co/api/v2/pokemon/1/
ivysaur → https://pokeapi.co/api/v2/pokemon/2/
venusaur → https://pokeapi.co/api/v2/pokemon/3/
...

Anatomía de una respuesta

r = requests.get("https://jsonplaceholder.typicode.com/posts/1", timeout=10)

print(r.status_code)  # 200
print(r.headers["Content-Type"])  # application/json; charset=utf-8
print(r.json())
# {'userId': 1, 'id': 1, 'title': '...', 'body': '...'}

4.3 Autenticación, tokens y variables de entorno

PokeAPI y JSONPlaceholder son abiertas, pero OpenWeather, GitHub, Twitter/X y casi toda API profesional exigen credencial. Hay dos patrones:

Patrón	Dónde va el secreto	Ejemplo
API Key en query	URL: ?appid=TU_KEY	OpenWeather
Bearer token en header	Authorization: Bearer TU_TOKEN	GitHub, la mayoría modernas

import requests

# Patrón 1: API Key en query (OpenWeather)
params = {"q": "Bogota", "appid": API_KEY, "units": "metric"}
r = requests.get("https://api.openweathermap.org/data/2.5/weather", params=params, timeout=10)

# Patrón 2: Bearer token en header (patrón recomendado)
headers = {"Authorization": f"Bearer {API_TOKEN}"}
r = requests.get("https://api.github.com/user/repos", headers=headers, timeout=10)

El secreto nunca va en el código

El error que reprueba la pre-entrega y que en trabajo real es un incidente de seguridad: escribir el token directamente en el .py y subirlo a GitHub.

# .env  (este archivo NUNCA se sube a git)
OPENWEATHER_KEY=abc123_tu_key_real
BASE_URL=https://api.openweathermap.org/data/2.5

# .gitignore (en la raíz del repo)
.env
__pycache__/
*.pyc
data_extracted.*

# script.py — solo referencia NOMBRES, nunca valores
import os
import requests
from dotenv import load_dotenv

load_dotenv()  # lee .env hacia el entorno del proceso

api_key = os.environ["OPENWEATHER_KEY"]  # KeyError claro si falta
base_url = os.environ.get("BASE_URL", "https://api.openweathermap.org/data/2.5")

params = {"q": "Bogota", "appid": api_key, "units": "metric"}
r = requests.get(f"{base_url}/weather", params=params, timeout=10)
r.raise_for_status()
print(r.json()["main"]["temp"])

Para instalar la librería: pip install python-dotenv requests pandas.

4.4 Estructuras JSON: mapas, listas y patrones anidados

JSON tiene solo 2 estructuras compuestas. Todo lo demás es combinación:

• Mapa / objeto → dict en Python: {"name": "ditto", "height": 3}
• Lista / arreglo → list en Python: [{"name": "a"}, {"name": "b"}]

import requests

ditto = requests.get("https://pokeapi.co/api/v2/pokemon/ditto", timeout=10).json()

print(type(ditto))              # <class 'dict'> → mapa raíz
print(ditto.keys())             # name, height, weight, stats, types, abilities...
print(type(ditto["stats"]))     # <class 'list'> → lista de mapas
print(ditto["stats"][0])
# {'base_stat': 48, 'effort': 0, 'stat': {'name': 'hp', 'url': '...'}}

Los 3 patrones que verás en el 95% de las APIs

Patrón A — Lista directa de objetos (JSONPlaceholder):

[
  {"userId": 1, "id": 1, "title": "..."},
  {"userId": 1, "id": 2, "title": "..."}
]

Patrón B — Mapa con metadata + lista de resultados (PokeAPI):

{
  "count": 1302,
  "next": "https://pokeapi.co/api/v2/pokemon?offset=20&limit=20",
  "previous": null,
  "results": [{"name": "bulbasaur", "url": "..."}]
}

Patrón C — Objeto con listas anidadas (detalle de un Pokémon):

{
  "name": "ditto",
  "types": [{"slot": 1, "type": {"name": "normal"}}],
  "stats": [{"base_stat": 48, "stat": {"name": "hp"}}]
}

Regla diagnóstica en una línea:

dato = respuesta.json()
print(type(dato))  # list → Patrón A | dict → mira sus claves: ¿hay 'results'/'data'? → Patrón B, ¿hay sublistas? → Patrón C

4.5 De JSON a tablas: estrategias y patrones

Un modelo no consume JSON anidado: consume filas y columnas. La conversión tiene 3 estrategias según el patrón:

Estrategia 1: lista plana → DataFrame directo

import pandas as pd
import requests

posts = requests.get("https://jsonplaceholder.typicode.com/posts", timeout=10).json()
df = pd.DataFrame(posts)

print(df.shape)   # (100, 4)
print(df.head(2))
print(df.dtypes)

Estrategia 2: seleccionar y aplanar campos del detalle

El listado de PokeAPI solo trae name + url. El dato analizable (altura, peso, experiencia) está en el detalle de cada URL. Hay que iterar:

import pandas as pd
import requests

lista = requests.get(
    "https://pokeapi.co/api/v2/pokemon", params={"limit": 10, "offset": 0}, timeout=10
).json()["results"]

filas = []
for item in lista:
    det = requests.get(item["url"], timeout=10).json()
    filas.append({
        "name": det["name"],
        "height": det["height"],
        "weight": det["weight"],
        "base_experience": det.get("base_experience"),  # .get: None si falta
        "n_types": len(det.get("types", [])),
        "n_abilities": len(det.get("abilities", [])),
    })

df = pd.DataFrame(filas)
print(df.head())

Estrategia 3: explotar una sublista (uno a muchos)

Un Pokémon tiene varios stats. Cada stat merece su propia fila conservando a qué Pokémon pertenece:

ditto = requests.get("https://pokeapi.co/api/v2/pokemon/ditto", timeout=10).json()

filas_stats = [
    {"pokemon": ditto["name"], "stat": s["stat"]["name"], "valor": s["base_stat"]}
    for s in ditto["stats"]
]
df_stats = pd.DataFrame(filas_stats)
print(df_stats)

  pokemon   stat  valor
0   ditto     hp     48
1   ditto attack     48
2   ditto defense     48
...

4.6 Normalización de JSON para análisis de datos

pandas.json_normalize aplana jerarquías automáticamente. Tres usos de menor a mayor potencia:

import pandas as pd

# 1. Aplana un dict con puntos (o el separador que elijas)
df = pd.json_normalize(ditto, sep="_")
print([c for c in df.columns if "stat" in c][:5])

# 2. Explotar una sublista conservando el contexto con meta
stats = pd.json_normalize(
    ditto,               # documento raíz (puede ser un dict o lista de dicts)
    record_path="stats", # sublista que se vuelve filas
    meta=["name", "height", "weight"],  # campos del padre que se repiten
    sep="_",
)
print(stats[["name", "stat_name", "base_stat"]].head())

# 3. Doble nivel: stats → stat → name, con separador legible
# stat_name viene de aplanar {"stat": {"name": "hp"}} con sep="_"

Para el Patrón B (lista dentro de un mapa), normaliza la lista interior:

data = requests.get(
    "https://pokeapi.co/api/v2/pokemon", params={"limit": 20, "offset": 0}, timeout=10
).json()

df = pd.json_normalize(data["results"])
print(df.head())

4.7 Resiliencia: manejo de errores, rate limits y reintentos

Un script de adquisición profesional asume que la red falla. Hay 4 capas de defensa, de adentro hacia afuera:

Capa 1: timeout + verificación de estado

import requests

try:
    r = requests.get("https://pokeapi.co/api/v2/pokemon/ditto", timeout=10)
    r.raise_for_status()
except requests.exceptions.Timeout:
    print("La API no respondió en 10s. Reintentando más tarde.")
except requests.exceptions.HTTPError as e:
    print(f"HTTP {r.status_code}: {e}")
except requests.exceptions.RequestException as e:
    print(f"Fallo de red: {e}")

Capa 2: validar el contenido, no solo el status

Un 200 con JSON inesperado rompe el pipeline dos etapas después. Falla temprano:

datos = r.json()

# Contrato mínimo: verifica forma antes de procesar
assert isinstance(datos, dict), f"Se esperaba dict, llegó {type(datos)}"
assert "name" in datos, "Falta clave 'name': la API cambió su schema"

Capa 3: detectar rate limit (429) y respetar Retry-After

if r.status_code == 429:
    espera = int(r.headers.get("Retry-After", 60))
    print(f"Límite excedido. Esperando {espera}s según el servidor...")

Capa 4: distinguir errores reintentables de definitivos

Reintentable (espera y prueba de nuevo)	Definitivo (registra y sigue)
Timeout, ConnectionError	401 (token inválido: reintentar no lo arregla)
429 (con espera)	404 (ese ítem no existe)
500, 502, 503	400 (tu petición está mal formada)

4.8 Estrategias de reintento y manejo de errores

Backoff exponencial: esperar cada vez más

Reintentar inmediatamente satura al servidor caído. El backoff espera 1s, 2s, 4s…:

import time
import requests

def get_con_reintentos(url, max_intentos=4, espera_base=1):
    for intento in range(1, max_intentos + 1):
        try:
            r = requests.get(url, timeout=10)
            if r.status_code == 429:
                espera = int(r.headers.get("Retry-After", espera_base * (2 ** intento)))
                print(f"429: esperando {espera}s (intento {intento})")
                time.sleep(espera)
                continue
            r.raise_for_status()
            return r.json()
        except (requests.exceptions.Timeout, requests.exceptions.ConnectionError) as e:
            if intento == max_intentos:
                raise
            espera = espera_base * (2 ** (intento - 1))
            print(f"Intento {intento} falló ({e}). Reintentando en {espera}s...")
            time.sleep(espera)

Sesión con reintentos automáticos (patrón profesional)

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

sesion = requests.Session()
reintentos = Retry(
    total=3,
    backoff_factor=1,          # espera 1s, 2s, 4s entre intentos
    status_forcelist=[429, 500, 502, 503, 504],
    allowed_methods=["GET"],   # solo reintenta lecturas seguras
)
sesion.mount("https://", HTTPAdapter(max_retries=reintentos))

r = sesion.get("https://pokeapi.co/api/v2/pokemon/ditto", timeout=10)
r.raise_for_status()

Paginación robusta con corte y registro

import requests

def extraer_pokemon(limit=20, max_paginas=5):
    url = "https://pokeapi.co/api/v2/pokemon"
    params = {"limit": limit, "offset": 0}
    todos, paginas = [], 0

    while url and paginas < max_paginas:
        r = requests.get(url, params=params, timeout=10)
        r.raise_for_status()
        data = r.json()

        todos.extend(data["results"])
        print(f"Página {paginas + 1}: {len(data['results'])} ítems (total: {len(todos)})")

        url = data["next"]  # PokeAPI devuelve la URL siguiente o null
        params = None       # la URL 'next' ya trae los params
        paginas += 1

    print(f"Extracción cerrada: {len(todos)} ítems en {paginas} páginas.")
    return todos

Script completo nivel pre-entrega: extraer → transformar → guardar

# extract_pokemon.py — pre-entrega M4: extracción segura y reproducible
import os
import json
from datetime import date, datetime
from pathlib import Path

import pandas as pd
import requests
from dotenv import load_dotenv

load_dotenv()
LIMIT = int(os.environ.get("LIMIT", 20))  # configuración, no secreto

def extraer(limit=LIMIT):
    r = requests.get(
        "https://pokeapi.co/api/v2/pokemon",
        params={"limit": limit, "offset": 0},
        timeout=10,
    )
    r.raise_for_status()
    return r.json()["results"]

def transformar(lista):
    filas = []
    for item in lista:
        try:
            det = requests.get(item["url"], timeout=10).json()
            filas.append({
                "name": det["name"],
                "height": det["height"],
                "weight": det["weight"],
                "base_experience": det.get("base_experience"),
                "n_types": len(det.get("types", [])),
                "fecha_extraccion": date.today().isoformat(),
            })
        except requests.exceptions.RequestException as e:
            print(f"Saltando {item['name']}: {e}")
    return pd.DataFrame(filas)

def guardar(df, destino=Path("data_extracted.csv")):
    df.to_csv(destino, index=False, encoding="utf-8")
    meta = {
        "fuente": "https://pokeapi.co/api/v2/pokemon",
        "fecha_extraccion": datetime.now().isoformat(),
        "n_registros": len(df),
        "limit": LIMIT,
    }
    destino.with_suffix(".meta.json").write_text(json.dumps(meta, indent=2), encoding="utf-8")
    print(f"Guardado: {destino} ({len(df)} filas) + metadata.")

if __name__ == "__main__":
    datos = extraer()
    df = transformar(datos)
    assert not df.empty, "DataFrame vacío: la extracción falló"
    guardar(df)

4.9 Buenas prácticas: reproducibilidad, documentación y control de versiones

Un extraction script sin reproducibilidad no es un entregable: es un accidente que funcionó una vez.

Reproducibilidad: 4 evidencias

1. requirements.txt  → requests, pandas, python-dotenv con versiones
2. Fecha de corte    → columna fecha_extraccion + .meta.json
3. Parámetros        → LIMIT y BASE_URL en .env, no hardcodeados
4. Validación        → assert df no vacío + contrato de columnas

# requirements.txt
requests==2.32.3
pandas==2.2.3
python-dotenv==1.0.1

# Contrato: falla temprano si la API cambió
requeridas = {"name", "height", "weight"}
faltantes = requeridas - set(df.columns)
if faltantes:
    raise ValueError(f"Schema inesperado, faltan: {faltantes}")

Documentación: README que permite repetir tu resultado

# Extracción Pokémon vía API (M4)

## Instalación
pip install -r requirements.txt

## Configuración
cp .env.example .env  # edita LIMIT si quieres más registros

## Ejecución
python extract_pokemon.py

## Salida
- data_extracted.csv (name, height, weight, base_experience, fecha_extraccion)
- data_extracted.meta.json (fuente, fecha, n_registros)

Control de versiones: qué se commitea y qué no

Se sube al repo	No se sube jamás
extract_pokemon.py, README.md, requirements.txt	.env (secretos y config local)
.gitignore, .env.example (plantilla sin valores)	__pycache__/, *.pyc
data_extracted.csv (lo pide la consigna)	Archivos gigantes o con datos sensibles

4.10 Preguntas de entrevista

4.11 Recursos recomendados

4.12 Test Yourself