How to create a transformation proxy for Data APIs: step by step

This tutorial shows how to build a transformation proxy for Data APIs: a small API that receives calls, queries an upstream API, processes/normalizes the JSON and returns an aggregated response. Useful for harmonizing schemas, reducing payloads and encapsulating transformation logic alongside calls.

Prerequisites

• Python 3.8+ installed
• pip to install packages (Flask, requests)
• Basic knowledge of JSON and REST
• Terminal and editor (VS Code, etc.)

Step 1: Why a transformation proxy?

A transformation proxy allows centralizing the logic that adapts multiple upstream APIs to the format your application needs. It prevents multiple consumers from applying transformations and eases maintenance. It can also reduce traffic by returning only the necessary fields.

Step 2: Create the environment and install dependencies

Create the project directory and a virtual environment. Install Flask and requests to create the API and make HTTP calls.

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

Step 3: Minimal proxy structure in Flask

Let's create a simple API that accepts GET on /proxy?query=... , calls the upstream API, transforms and returns JSON. We keep the code minimal and commented to be didactic.

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)

Step 4: Handle errors and missing fields

Upstream APIs can return incomplete responses. Adding simple validations and default values prevents crashes. In the example we use .get() and default values; you can also log errors and return appropriate HTTP codes.

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

Step 5: Filter fields and reduce payload

To optimize traffic, the proxy should return only the necessary fields. In transform_data we already chose three fields. You can add query parameters to request more/less fields.

@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

Step 6: Simple in-memory cache (optional)

To reduce repeated calls, you can use an in-memory cache with a short TTL. Useful in development environments. In production use Redis or 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

Verify the result

Run the API locally and test with curl or a browser. You should receive JSON with count, avg_score and items. Examples of tests and common errors:

• Command: curl "http://localhost:5000/proxy?query=test" — returns transformed JSON.
• 502 error: check the URL and access to the upstream API and timeout.
• Response with count 0: the upstream returned a different structure; check logs or print resp.json().

Conclusion

A transformation proxy in Flask is a practical way to unify and reduce complexity in integrations with Data APIs. Next steps: add automated tests, authentication (API keys), and replace the in-memory cache with Redis. Tip: which specific transformation do you want to centralize first in your architecture?