Cómo filtrar y agregar datos de una API REST en Python: paso a paso

Este tutorial muestra cómo filtrar y agregar datos recibidos de una API REST en Python para obtener informes resumidos y analizables. Saber filtrar y agregar directamente en la capa cliente es útil cuando la API no proporciona endpoints analíticos o cuando se quiere validar datos antes de cargar a un sistema ETL.

Requisitos previos

• Python 3.8+ instalado.
• Bibliotecas: requests y pandas (pip install requests pandas).
• Una API REST que devuelva JSON con registros similares (p. ej.: listado de transacciones o eventos).
• Editor de texto/IDE y terminal.

Paso 1: comprender el formato de la API y elegir filtros

Antes de escribir código, consulte la documentación de la API: parámetros de query, campos devueltos y paginación. Decida los filtros que va a aplicar (fechas, estado, tipo) y las claves para agregar (p. ej.: categoría, día, cliente).

Paso 2: obtener páginas de la API con requests

Implemente la llamada a la API con paginación. Ejemplo mínimo: request por página hasta que no haya más resultados. Gestione errores HTTP y tiempos de espera.

import requests

def fetch_all(url, params=None, headers=None, page_param='page'):
    params = params.copy() if params else {}
    page = 1
    all_items = []
    while True:
        params[page_param] = page
        resp = requests.get(url, params=params, headers=headers, timeout=10)
        resp.raise_for_status()
        data = resp.json()
        items = data.get('results') or data.get('items') or data
        if not items:
            break
        all_items.extend(items)
        # condição simples de paragem: quando menos itens que uma página típica
        if isinstance(items, list) and len(items) < 100:
            break
        page += 1
    return all_items

Paso 3: aplicar filtros en el cliente

Filtrar en el cliente es útil cuando la API no acepta todos los parámetros. Aplique filtros de forma clara y reproducible. Ejemplo: filtrar por fecha y estado.

from datetime import datetime

def filter_items(items, start_date=None, end_date=None, status=None):
    def in_range(d):
        if not d:
            return False
        dt = datetime.fromisoformat(d)
        if start_date and dt < start_date:
            return False
        if end_date and dt > end_date:
            return False
        return True

    filtered = []
    for it in items:
        if status and it.get('status') != status:
            continue
        if start_date or end_date:
            if not in_range(it.get('created_at')):
                continue
        filtered.append(it)
    return filtered

Paso 4: transformar JSON a pandas DataFrame

Convertir a DataFrame facilita la agregación y limpieza. Normalice listas anidadas o campos opcionales antes de agregar.

import pandas as pd

def to_dataframe(items):
    df = pd.json_normalize(items)
    # converter colunas de data
    if 'created_at' in df.columns:
        df['created_at'] = pd.to_datetime(df['created_at'])
    return df

Paso 5: agregar y calcular métricas

Use groupby para resumir por clave. Ejemplo: sumar importes por día y por categoría, contar transacciones y calcular la media.

def aggregate_metrics(df):
    # criar coluna dia
    df['day'] = df['created_at'].dt.date
    # exemplo de métricas por dia e categoria
    agg = (df.groupby(['day', 'category'])
             .agg(total_amount=pd.NamedAgg(column='amount', aggfunc='sum'),
                  count=pd.NamedAgg(column='id', aggfunc='count'),
                  avg_amount=pd.NamedAgg(column='amount', aggfunc='mean'))
             .reset_index())
    return agg

Paso 6: tratar errores comunes y rendimiento

Errores comunes: JSON inesperado, timeout, límites de tasa y campos ausentes. Para grandes volúmenes, pagine con parámetros de la API, procese por bloques y evite cargar todo en memoria. Use chunks y escriba resultados intermedios a archivos CSV/Parquet.

# exemplo simples de escrita por chunks
for i in range(0, len(all_items), 1000):
    chunk = all_items[i:i+1000]
    df_chunk = to_dataframe(chunk)
    df_chunk.to_parquet(f'data_chunk_{i//1000}.parquet')

Verificar el resultado

Abra el DataFrame agregado (agg) y confirme sumas, recuentos y fechas. Compruebe algunas filas sin procesar para confirmar que los filtros funcionaron. Ejemplos de verificación rápida:

print(agg.head())
print(df[['id','created_at','status','amount']].sample(5))
print('Total original:', len(all_items), 'Após filtros:', len(df))

Conclusión

Filtrar y agregar datos de una API REST en Python permite crear informes incluso cuando la API no dispone de endpoints analíticos. Próximos pasos: añadir autenticación (Bearer/OAuth), usar herramientas como Dask para grandes volúmenes o automatizar como parte de un pipeline ETL. Consejo: pruebe a almacenar salidas en Parquet para consultas más rápidas — ¿qué agregación tendría sentido para su equipo?