Cómo crear un proxy de transformación en APIs de Datos: paso a paso

Este tutorial muestra cómo construir un proxy de transformación para APIs de Datos: una pequeña API que recibe llamadas, consulta una API upstream, procesa/normaliza los JSON y devuelve una respuesta agregada. Útil para armonizar esquemas, reducir payloads y encapsular lógica de transformación junto con las llamadas.

Requisitos previos

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

Paso 1: ¿Por qué un proxy de transformación?

Un proxy de transformación permite centralizar la lógica que adapta varias APIs upstream al formato que su aplicación necesita. Evita múltiples consumidores aplicando transformaciones y facilita el mantenimiento. También puede reducir tráfico al devolver solo los campos necesarios.

Paso 2: Crear el entorno e instalar dependencias

Crear un directorio del proyecto y un entorno virtual. Instalar Flask y requests para crear la API y hacer llamadas HTTP.

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

Paso 3: Estructura mínima del proxy en Flask

Vamos a crear una API simple que acepta GET en /proxy?query=... , llama a la API upstream, transforma y devuelve JSON. Mantenemos el código mínimo y comentado para que sea didáctico.

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)

Paso 4: Manejar errores y campos faltantes

Las APIs upstream pueden devolver respuestas incompletas. Añadir validaciones simples y valores por omisión evita crashes. En el ejemplo usamos .get() y valores por omisión; también puede registrar errores y devolver códigos HTTP apropiados.

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

Paso 5: Filtrar campos y reducir payload

Para optimizar tráfico, el proxy debe devolver solo los campos necesarios. En transform_data ya hemos escogido tres campos. Puede añadir parámetros query para pedir más/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 respetar fields

Paso 6: Caché simple en memoria (opcional)

Para reducir llamadas repetidas, puede usar una caché en memoria con TTL corto. Útil en entornos de desarrollo. En producción use Redis o 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 el resultado

Ejecutar la API localmente y probar con curl o el navegador. Debe recibir JSON con count, avg_score e items. Ejemplos de pruebas y errores comunes:

• Comando: curl "http://localhost:5000/proxy?query=test" — devuelve JSON transformado.
• Error 502: verificar URL y acceso a la API upstream y tiempo de espera.
• Respuesta con count 0: la upstream devolvió una estructura diferente; ver logs o imprimir resp.json().

Conclusión

Un proxy de transformación en Flask es una forma práctica de unificar y reducir la complejidad en las integraciones con APIs de Datos. Pasos siguientes: añadir pruebas automatizadas, autenticación (API keys), y sustituir la caché en memoria por Redis. Consejo: ¿qué transformación específica quiere centralizar primero en su arquitectura?