Tooldata Academia
Lección 23 de 29•Módulo 5: Web Scraping y APIs

23. Consumo de APIs REST

Requests, autenticación, paginación y manejo de errores

25 minutos

Aprende a consumir APIs REST profesionalmente para extraer datos de servicios externos. Las APIs son la forma más confiable y ética de obtener datos estructurados de plataformas como Twitter, LinkedIn, Stripe y miles más.

¿Qué es una API REST?

API (Application Programming Interface): Una interfaz que permite a aplicaciones comunicarse entre sí.

REST (Representational State Transfer): Un estilo de arquitectura para APIs que usa HTTP.

¿Por qué usar APIs en lugar de scraping?

  • Datos estructurados (JSON/XML)
  • Más rápido y confiable
  • Respaldado oficialmente por la plataforma
  • Documentación completa
  • Rate limits claros

Ejemplo real: En lugar de scrapear perfiles de LinkedIn (prohibido), usas su API oficial para obtener datos de empresas y empleos.

Fundamentos de HTTP

Métodos HTTP principales

import requests
import json

# GET - Obtener datos
response = requests.get('https://api.github.com/users/github')
print(f"Status: {response.status_code}")
print(f"Usuario: {response.json()['login']}")

# POST - Crear recursos
data = {'name': 'Nuevo repositorio', 'private': False}
# response = requests.post('https://api.github.com/user/repos', json=data, headers=headers)

# PUT - Actualizar completo
# PATCH - Actualizar parcial
# DELETE - Eliminar

# Códigos de status comunes:
# 200 OK - Éxito
# 201 Created - Recurso creado
# 400 Bad Request - Request inválido
# 401 Unauthorized - Sin autenticación
# 403 Forbidden - Sin permisos
# 404 Not Found - Recurso no existe
# 429 Too Many Requests - Excediste rate limit
# 500 Server Error - Error del servidor

Tu primera API: GitHub

Ejemplo básico sin autenticación

import requests
import pandas as pd

# GitHub API (pública, no requiere token para requests limitados)
username = 'torvalds'  # Linus Torvalds
url = f'https://api.github.com/users/{username}'

response = requests.get(url)

if response.status_code == 200:
    user = response.json()
    print(f"Nombre: {user['name']}")
    print(f"Bio: {user['bio']}")
    print(f"Repos públicos: {user['public_repos']}")
    print(f"Followers: {user['followers']}")
else:
    print(f"Error: {response.status_code}")

# Obtener repositorios del usuario
repos_url = f'https://api.github.com/users/{username}/repos'
repos_response = requests.get(repos_url)
repos = repos_response.json()

# Convertir a DataFrame
df_repos = pd.DataFrame(repos)
print(f"\nRepositorios: {len(df_repos)}")
print(df_repos[['name', 'stargazers_count', 'language']].head(10))

# Análisis
print(f"\nTop 5 repos por estrellas:")
top_repos = df_repos.nlargest(5, 'stargazers_count')[['name', 'stargazers_count', 'language']]
print(top_repos)

Autenticación

API Keys y Bearer Tokens

# Método 1: API Key en query params
api_key = 'TU_API_KEY_AQUI'
url = f'https://api.example.com/data?api_key={api_key}'
response = requests.get(url)

# Método 2: Bearer Token en headers (más seguro)
headers = {
    'Authorization': f'Bearer {api_key}',
    'Accept': 'application/json'
}
response = requests.get('https://api.example.com/data', headers=headers)

# Método 3: Basic Auth (usuario:contraseña)
from requests.auth import HTTPBasicAuth
response = requests.get(
    'https://api.example.com/data',
    auth=HTTPBasicAuth('username', 'password')
)

# Método 4: OAuth 2.0 (usado por Google, Facebook, Twitter)
# Requiere flujo de autorización completo
# Usualmente se usa librería específica (tweepy, google-api-python-client)

Manejo seguro de credenciales

import os
from dotenv import load_dotenv

# Crear archivo .env (NO commitear a git!)
# API_KEY=tu_key_secreta_aqui
# API_SECRET=tu_secret_aqui

# Cargar variables de entorno
load_dotenv()

API_KEY = os.getenv('API_KEY')
API_SECRET = os.getenv('API_SECRET')

if not API_KEY:
    raise ValueError("API_KEY no encontrada. Verifica tu archivo .env")

headers = {'Authorization': f'Bearer {API_KEY}'}

Paginación

Tipos de paginación

import requests
import time

# TIPO 1: Offset/Limit (page-based)
def fetch_with_offset(base_url, total_items=100):
    """Paginación por offset"""
    all_data = []
    limit = 20
    offset = 0

    while offset < total_items:
        response = requests.get(f"{base_url}?limit={limit}&offset={offset}")
        data = response.json()

        if not data:  # No más datos
            break

        all_data.extend(data)
        offset += limit
        time.sleep(0.5)  # Rate limiting

    return all_data

# TIPO 2: Cursor-based (más eficiente para grandes datasets)
def fetch_with_cursor(base_url):
    """Paginación por cursor"""
    all_data = []
    cursor = None

    while True:
        params = {'cursor': cursor} if cursor else {}
        response = requests.get(base_url, params=params)
        result = response.json()

        all_data.extend(result['data'])

        cursor = result.get('next_cursor')
        if not cursor:  # No hay más páginas
            break

        time.sleep(0.5)

    return all_data

# TIPO 3: Link header (usado por GitHub)
def fetch_with_link_header(url, headers):
    """Paginación usando Link header"""
    all_data = []

    while url:
        response = requests.get(url, headers=headers)
        all_data.extend(response.json())

        # GitHub devuelve siguiente página en header 'Link'
        link_header = response.headers.get('Link', '')
        if 'rel="next"' in link_header:
            # Parsear URL del siguiente link
            url = link_header.split(';')[0].strip('<>')
        else:
            url = None

        time.sleep(1)

    return all_data

Rate Limiting

Respetando límites de APIs

import time
from datetime import datetime, timedelta

class APIRateLimiter:
    """Maneja rate limiting inteligente"""
    def __init__(self, calls_per_hour=1000):
        self.calls_per_hour = calls_per_hour
        self.calls = []

    def wait_if_needed(self):
        """Espera si excediste el límite"""
        now = datetime.now()
        hour_ago = now - timedelta(hours=1)

        # Eliminar calls antiguas (más de 1 hora)
        self.calls = [call for call in self.calls if call > hour_ago]

        if len(self.calls) >= self.calls_per_hour:
            # Calcular cuánto esperar
            oldest_call = min(self.calls)
            wait_until = oldest_call + timedelta(hours=1)
            wait_seconds = (wait_until - now).total_seconds()

            if wait_seconds > 0:
                print(f"Rate limit alcanzado. Esperando {wait_seconds:.0f}s...")
                time.sleep(wait_seconds)

        self.calls.append(now)

# Uso
limiter = APIRateLimiter(calls_per_hour=100)

for i in range(150):
    limiter.wait_if_needed()
    # response = requests.get(url)
    print(f"Request #{i+1}")

Leer rate limits de headers

def check_rate_limit(response):
    """Lee información de rate limit de headers"""
    headers = response.headers

    # GitHub y muchas APIs usan estos headers
    limit = headers.get('X-RateLimit-Limit')
    remaining = headers.get('X-RateLimit-Remaining')
    reset_timestamp = headers.get('X-RateLimit-Reset')

    if all([limit, remaining, reset_timestamp]):
        reset_time = datetime.fromtimestamp(int(reset_timestamp))
        print(f"Rate limit: {remaining}/{limit}")
        print(f"Reset: {reset_time.strftime('%H:%M:%S')}")

        if int(remaining) < 10:
            print("⚠️ ALERTA: Quedan pocas requests!")
            return False

    return True

# Uso
response = requests.get('https://api.github.com/users/github')
check_rate_limit(response)

Manejo de errores robusto

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

def create_session_with_retries():
    """Crea sesión con reintentos automáticos"""
    session = requests.Session()

    # Configurar estrategia de reintentos
    retry_strategy = Retry(
        total=3,  # 3 reintentos
        backoff_factor=1,  # Esperar 1s, 2s, 4s entre reintentos
        status_forcelist=[429, 500, 502, 503, 504],  # Reintentar en estos códigos
        allowed_methods=["HEAD", "GET", "OPTIONS"]  # Solo métodos seguros
    )

    adapter = HTTPAdapter(max_retries=retry_strategy)
    session.mount("http://", adapter)
    session.mount("https://", adapter)

    return session

# Uso
session = create_session_with_retries()

try:
    response = session.get('https://api.example.com/data', timeout=10)
    response.raise_for_status()  # Lanza excepción si status no es 2xx
    data = response.json()
    print(f"Datos recibidos: {len(data)} items")

except requests.exceptions.HTTPError as e:
    print(f"Error HTTP: {e}")
except requests.exceptions.ConnectionError:
    print("Error de conexión")
except requests.exceptions.Timeout:
    print("Timeout - servidor muy lento")
except requests.exceptions.RequestException as e:
    print(f"Error inesperado: {e}")

Proyecto práctico: API de CoinGecko (Crypto)

import requests
import pandas as pd
import time

class CryptoAPI:
    """Cliente para API de CoinGecko (gratis, sin auth)"""
    BASE_URL = "https://api.coingecko.com/api/v3"

    def __init__(self):
        self.session = requests.Session()

    def get_top_coins(self, limit=50):
        """Obtiene top criptomonedas por market cap"""
        endpoint = f"{self.BASE_URL}/coins/markets"
        params = {
            'vs_currency': 'usd',
            'order': 'market_cap_desc',
            'per_page': limit,
            'page': 1,
            'sparkline': False
        }

        response = self.session.get(endpoint, params=params)
        response.raise_for_status()

        coins = response.json()
        return pd.DataFrame(coins)

    def get_coin_history(self, coin_id, days=30):
        """Obtiene histórico de precios"""
        endpoint = f"{self.BASE_URL}/coins/{coin_id}/market_chart"
        params = {
            'vs_currency': 'usd',
            'days': days
        }

        response = self.session.get(endpoint, params=params)
        response.raise_for_status()

        data = response.json()
        prices = data['prices']  # Lista de [timestamp, precio]

        df = pd.DataFrame(prices, columns=['timestamp', 'price'])
        df['date'] = pd.to_datetime(df['timestamp'], unit='ms')
        return df[['date', 'price']]

# Usar el cliente
api = CryptoAPI()

# Top 20 cryptos
print("Obteniendo top 20 criptomonedas...")
df_coins = api.get_top_coins(limit=20)

print(f"\nTop 20 cryptos por market cap:")
print(df_coins[['name', 'symbol', 'current_price', 'market_cap', 'price_change_percentage_24h']].head(10))

# Análisis
print(f"\nAnálisis 24h:")
print(f"Mayor ganancia: {df_coins.loc[df_coins['price_change_percentage_24h'].idxmax(), 'name']} "
      f"({df_coins['price_change_percentage_24h'].max():.2f}%)")
print(f"Mayor pérdida: {df_coins.loc[df_coins['price_change_percentage_24h'].idxmin(), 'name']} "
      f"({df_coins['price_change_percentage_24h'].min():.2f}%)")

# Histórico de Bitcoin
print("\nObteniendo histórico de Bitcoin...")
time.sleep(1)  # Rate limiting
df_btc = api.get_coin_history('bitcoin', days=30)

print(f"\nBitcoin últimos 30 días:")
print(f"Precio actual: ${df_btc.iloc[-1]['price']:,.2f}")
print(f"Precio hace 30 días: ${df_btc.iloc[0]['price']:,.2f}")
change = ((df_btc.iloc[-1]['price'] - df_btc.iloc[0]['price']) / df_btc.iloc[0]['price']) * 100
print(f"Cambio: {change:+.2f}%")

# Guardar datos
df_coins.to_csv('top_cryptos.csv', index=False)
df_btc.to_csv('bitcoin_30days.csv', index=False)
print("\nDatos guardados en CSV")

Resumen de la lección

✅ APIs REST son la forma profesional de obtener datos estructurados ✅ Autenticación: API keys, Bearer tokens, OAuth 2.0 ✅ Paginación: Offset, cursor, link headers ✅ Rate limiting: Respetar límites para evitar bloqueos ✅ Manejo de errores: Reintentos automáticos, timeouts ✅ Session objects: Reutilizar conexiones para mejor performance ✅ Siempre lee la documentación oficial de la API

Checklist para consumir APIs

  • Lee la documentación completa
  • Guarda credenciales en .env (no en código)
  • Implementa rate limiting
  • Maneja paginación correctamente
  • Implementa reintentos para errores transitorios
  • Usa timeouts en todos los requests
  • Loggea errores para debugging
  • Cachea resultados cuando sea posible

🎯 Próxima lección: 24. Scraping de Redes Sociales

Aprenderás a extraer datos de redes sociales usando APIs oficiales (Twitter, LinkedIn) y técnicas avanzadas. ¡Nos vemos! 🚀

¿Completaste esta lección?

Marca esta lección como completada. Tu progreso se guardará en tu navegador.

Anterior
22. Introducción a Web Scraping con BeautifulSoup
Siguiente
24. Scraping de Redes Sociales