Como sincronizar dados via Webhook em APIs de Dados: passo a passo

Este tutorial mostra como receber e processar notificações com Webhook em APIs de Dados para sincronizar actualizações quase em tempo real. É útil para reduzir polling, reagir a eventos e manter uma cópia local dos dados actualizados.

Pré-requisitos

• Python 3.9+ instalado
• Bibliotecas: FastAPI, uvicorn, requests, pydantic, sqlite3 (builtin)
• Conhecimentos básicos de HTTP, JSON e de linhas de comando

Passo 1: Entender o cenário e escolher o fluxo

Vamos criar um receptor Webhook simples com FastAPI que valida o payload, verifica um segredo (HMAC) e actualiza uma tabela SQLite. O fluxo típico: o emissor envia um POST JSON; o receptor valida e responde 200; em caso de sucesso actualiza dados locais.

Passo 2: Estruturar o projecto e instalar dependências

Cria uma pasta para o projecto e instala as bibliotecas necessárias.

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

Passo 3: Criar a base de dados SQLite minimal

Criamos uma tabela simples para guardar registos recebidos. Isto facilita verificar se a sincronização está a funcionar.

-- 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()

Passo 4: Implementar o receptor Webhook com validação HMAC

Usamos HMAC-SHA256 para garantir que o pedido vem de uma fonte conhecida. O emissor e o receptor partilham um segredo. O emissor inclui um header X-Signature com o HMAC do 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 em produção

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'}

Passo 5: Testar localmente o Webhook (emissor simulado)

Simulamos o emissor com requests. Geramos o HMAC do body e enviamos o POST para o receptor local em 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)

Passo 6: Executar o receptor

Arranca o servidor FastAPI com uvicorn e depois executa send_test.py para enviar o evento.

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

Verificar o resultado

Confirma que recebes 200 OK na consola do emissor. Abre a base de dados SQLite e verifica a tabela items: o registo com id 'item-123' deve existir com data_json contendo os 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()

Conclusão

Agora tens um receptor Webhook funcional que valida HMAC, parseia JSON com Pydantic e actualiza uma base de dados local. Próximos passos: armazenar o SECRET em variáveis de ambiente, adicionar logging, lidar com retries e processar eventos em background (fila). Dica: como vais testar exposição pública, experimenta usar ngrok para receber Webhooks externos com segurança.