Como criar um proxy de transformação em APIs de Dados: passo a passo

Este tutorial mostra como construir um proxy de transformação para APIs de Dados: uma pequena API que recebe chamadas, consulta uma API upstream, processa/normaliza os JSON e devolve uma resposta agregada. Útil para harmonizar esquemas, reduzir payloads e encapsular lógica de transformação junto das chamadas.

Pré-requisitos

• Python 3.8+ instalado
• pip para instalar pacotes (Flask, requests)
• Conhecimentos básicos de JSON e REST
• Terminal e editor (VS Code, etc.)

Passo 1: Porquê um proxy de transformação?

Um proxy de transformação permite centralizar a lógica que adapta várias APIs upstream ao formato que a sua aplicação precisa. Evita múltiplos consumidores a aplicar transformações e facilita a manutenção. Também pode reduzir tráfego ao devolver apenas os campos necessários.

Passo 2: Criar o ambiente e instalar dependências

Criar um diretório do projeto e um ambiente virtual. Instalar Flask e requests para criar a API e fazer chamadas HTTP.

python -m venv venv
source venv/bin/activate   # Windows: venv\Scripts\activate
pip install Flask requests

Passo 3: Estrutura mínima do proxy em Flask

Vamos criar uma API simples que aceita GET em /proxy?query=... , chama a API upstream, transforma e devolve JSON. Mantemos o código mínimo e comentado para ser didático.

from flask import Flask, request, jsonify
import requests

app = Flask(__name__)
UPSTREAM_URL = 'https://api.exemplo.com/data'  # substituir pela API real

@app.route('/proxy')
def proxy():
    q = request.args.get('query', '')
    # Chamada à API upstream com timeout e tratamento básico de erros
    try:
        resp = requests.get(UPSTREAM_URL, params={'q': q}, timeout=5)
        resp.raise_for_status()
    except requests.RequestException as e:
        return jsonify({'error': 'upstream_error', 'detail': str(e)}), 502

    data = resp.json()
    # Transformação: normalizar e agregar
    transformed = transform_data(data)
    return jsonify(transformed)

def transform_data(data):
    # Exemplo: da estrutura upstream -> queremos uma lista de {id, name, score}
    items = []
    for r in data.get('results', []):
        item = {
            'id': r.get('id') or r.get('uid'),
            'name': r.get('title') or r.get('name'),
            'score': r.get('metrics', {}).get('score', 0)
        }
        items.append(item)
    # Agregação simples: count e média de score
    count = len(items)
    avg_score = sum(i['score'] for i in items) / count if count else 0
    return {'count': count, 'avg_score': avg_score, 'items': items}

if __name__ == '__main__':
    app.run(debug=True, port=5000)

Passo 4: Lidar com erros e campos em falta

As APIs upstream podem devolver respostas incompletas. Adicionar validações simples e valores por omissão evita crashes. No exemplo usamos .get() e valores por omissão; pode também registar erros e devolver códigos HTTP apropriados.

def transform_data(data):
    if not isinstance(data, dict):
        return {'count': 0, 'avg_score': 0, 'items': []}
    # resto igual ao exemplo anterior...

Passo 5: Filtrar campos e reduzir payload

Para optimizar tráfego, o proxy deve devolver apenas os campos necessários. No transform_data já escolhemos três campos. Pode adicionar parâmetros query para pedir mais/menos campos.

@app.route('/proxy')
def proxy():
    q = request.args.get('query', '')
    fields = request.args.get('fields', 'id,name,score').split(',')
    # chamada upstream como antes...
    data = requests.get(...).json()
    transformed = transform_data(data, fields)
    return jsonify(transformed)

# adaptar transform_data para respeitar fields

Passo 6: Cache simples em memória (opcional)

Para reduzir chamadas repetidas, pode usar um cache em memória com TTL curto. Útil em ambientes de desenvolvimento. Em produção usar Redis ou similar.

from time import time
CACHE = {}
TTL = 30  # segundos

def get_cached(key):
    entry = CACHE.get(key)
    if not entry: return None
    ts, value = entry
    if time() - ts > TTL:
        del CACHE[key]
        return None
    return value

# no proxy: key = f"{q}:{','.join(fields)}"; checar cache antes de requests.get

Verificar o resultado

Executar a API localmente e testar com curl ou browser. Deve receber JSON com count, avg_score e items. Exemplos de testes e erros comuns:

• Comando: curl "http://localhost:5000/proxy?query=test" — devolve JSON transformado.
• Erro 502: verificar URL e acesso à API upstream e tempo de espera.
• Resposta com count 0: a upstream devolveu estrutura diferente; ver logs ou imprimir resp.json().

Conclusão

Um proxy de transformação em Flask é uma forma prática de unificar e reduzir complexidade nas integrações com APIs de Dados. Próximos passos: adicionar testes automatizados, autenticação (API keys), e substituir cache em memória por Redis. Dica: que transformação específica quer centralizar primeiro na sua arquitetura?