Como filtrar e agregar dados de uma API REST em Python: passo a passo

Este tutorial mostra como filtrar e agregar dados recebidos de uma API REST em Python para obter relatórios resumidos e analisáveis. Saber filtrar e agregar diretamente na camada cliente é útil quando a API não fornece endpoints analíticos ou quando se quer validar dados antes de carregar para um sistema ETL.

Pré-requisitos

• Python 3.8+ instalado.
• Bibliotecas: requests e pandas (pip install requests pandas).
• Uma API REST que devolva JSON com registos semelhantes (ex.: listagem de transacções ou eventos).
• Editor de texto/IDE e terminal.

Passo 1: perceber o formato da API e escolher filtros

Antes de escrever código, verifique a documentação da API: parâmetros de query, campos devolvidos e paginação. Decida os filtros que vai aplicar (datas, estado, tipo) e as chaves para agregar (ex.: categoria, dia, cliente).

Passo 2: obter páginas da API com requests

Implemente a chamada à API com paginação. Exemplo mínimo: request por página até não haver mais resultados. Trate erros HTTP e tempos 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

Passo 3: aplicar filtros no cliente

Filtrar no cliente é útil quando a API não aceita todos os parâmetros. Aplique filtros de forma clara e reproduzível. Exemplo: filtrar por data e 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

Passo 4: transformar JSON para pandas DataFrame

Converter para DataFrame facilita agregação e limpeza. Normalize listas aninhadas ou campos opcionais 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

Passo 5: agregar e calcular métricas

Use groupby para sumarizar por chave. Exemplo: somar montantes por dia e por categoria, contar transacções e calcular média.

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

Passo 6: tratar erros comuns e desempenho

Erros comuns: JSON inesperado, timeout, limites de taxa e campos em falta. Para grandes volumes, pagine com parâmetros da API, processe por blocos e evite carregar tudo para memória. Use chunks e escreva resultados intermédios para ficheiros 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 o resultado

Abra o DataFrame agregado (agg) e confirme somas, contagens e datas. Verifique algumas linhas brutas para confirmar que os filtros funcionaram. Exemplos de verificação 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))

Conclusão

Filtrar e agregar dados de uma API REST em Python permite criar relatórios mesmo quando a API não tem endpoints analíticos. Próximos passos: adicionar autenticação (Bearer/OAuth), usar ferramentas como Dask para grandes volumes ou automatizar como parte de um pipeline ETL. Dica: experimente armazenar saídas em Parquet para consultas mais rápidas — que agregação faria sentido para a sua equipa?