Cómo sincronizar datos vía Webhook en APIs de Datos: paso a paso

Este tutorial muestra cómo recibir y procesar notificaciones con Webhook en APIs de Datos para sincronizar actualizaciones casi en tiempo real. Es útil para reducir polling, reaccionar a eventos y mantener una copia local de los datos actualizada.

Pre-requisitos

• Python 3.9+ instalado
• Bibliotecas: FastAPI, uvicorn, requests, pydantic, sqlite3 (builtin)
• Conocimientos básicos de HTTP, JSON y de líneas de comando

Paso 1: Entender el escenario y elegir el flujo

Vamos a crear un receptor Webhook simple con FastAPI que valida el payload, verifica un secreto (HMAC) y actualiza una tabla SQLite. El flujo típico: el emisor envía un POST JSON; el receptor valida y responde 200; en caso de éxito actualiza datos locales.

Paso 2: Estructurar el proyecto e instalar dependencias

Crea una carpeta para el proyecto e instala las bibliotecas necesarias.

mkdir webhook-sync
cd webhook-sync
python -m venv .venv
source .venv/bin/activate   # ou .venv\Scripts\activate no Windows
pip install fastapi uvicorn requests pydantic

Paso 3: Crear la base de datos SQLite minimal

Crearemos una tabla simple para guardar registros recibidos. Esto facilita comprobar si la sincronización está funcionando.

-- cria_db.sql
CREATE TABLE IF NOT EXISTS items (
    id TEXT PRIMARY KEY,
    data_json TEXT,
    updated_at TEXT
);

# executar no Python interativo
import sqlite3
conn = sqlite3.connect('data.db')
with open('cria_db.sql') as f:
    conn.executescript(f.read())
conn.commit()
conn.close()

Paso 4: Implementar el receptor Webhook con validación HMAC

Usamos HMAC-SHA256 para garantizar que la petición proviene de una fuente conocida. El emisor y el receptor comparten un secreto. El emisor incluye un header X-Signature con el HMAC del body.

# app.py
from fastapi import FastAPI, Request, Header, HTTPException
from pydantic import BaseModel
import hmac, hashlib, sqlite3, json, datetime

app = FastAPI()
SECRET = b'super-segredo-compartilhado'  # guardar de forma segura en producción

class Payload(BaseModel):
    id: str
    attributes: dict

def verify_hmac(body: bytes, signature: str) -> bool:
    mac = hmac.new(SECRET, body, hashlib.sha256).hexdigest()
    return hmac.compare_digest(mac, signature)

def upsert_item(item_id: str, data: dict):
    conn = sqlite3.connect('data.db')
    cur = conn.cursor()
    cur.execute(
        'REPLACE INTO items (id, data_json, updated_at) VALUES (?, ?, ?)',
        (item_id, json.dumps(data), datetime.datetime.utcnow().isoformat())
    )
    conn.commit()
    conn.close()

@app.post('/webhook')
async def webhook(request: Request, x_signature: str | None = Header(None)):
    body = await request.body()
    if not x_signature or not verify_hmac(body, x_signature):
        raise HTTPException(status_code=401, detail='Invalid signature')
    try:
        payload = Payload.parse_raw(body)
    except Exception:
        raise HTTPException(status_code=400, detail='Invalid payload')
    upsert_item(payload.id, payload.attributes)
    return {'status': 'ok'}

Paso 5: Probar localmente el Webhook (emisor simulado)

Simulamos el emisor con requests. Generamos el HMAC del body y enviamos el POST al receptor local en http://localhost:8000/webhook.

# send_test.py
import requests, hmac, hashlib, json

SECRET = b'super-segredo-compartilhado'
url = 'http://localhost:8000/webhook'
body = json.dumps({
    'id': 'item-123',
    'attributes': {'name':'Produto A','price':10.5}
}).encode()
signature = hmac.new(SECRET, body, hashlib.sha256).hexdigest()
headers = {'Content-Type': 'application/json', 'X-Signature': signature}
resp = requests.post(url, data=body, headers=headers)
print(resp.status_code, resp.text)

Paso 6: Ejecutar el receptor

Arranca el servidor FastAPI con uvicorn y después ejecuta send_test.py para enviar el evento.

uvicorn app:app --reload --port 8000
python send_test.py

Verificar el resultado

Confirma que recibes 200 OK en la consola del emisor. Abre la base de datos SQLite y verifica la tabla items: el registro con id 'item-123' debe existir con data_json conteniendo los atributos.

python
import sqlite3, json
conn = sqlite3.connect('data.db')
cur = conn.cursor()
cur.execute("SELECT id, data_json, updated_at FROM items WHERE id='item-123'")
row = cur.fetchone()
print(row)
conn.close()

Conclusión

Ahora tienes un receptor Webhook funcional que valida HMAC, parsea JSON con Pydantic y actualiza una base de datos local. Próximos pasos: almacenar el SECRET en variables de entorno, añadir logging, manejar retries y procesar eventos en background (cola). Consejo: si vas a exponerlo públicamente, prueba usar ngrok para recibir Webhooks externos con seguridad.