How to synchronize data via Webhook in Data APIs: step by step

This tutorial shows how to receive and process notifications with Webhook in Data APIs to synchronize near real-time updates. It is useful to reduce polling, react to events and keep a local copy of up-to-date data.

Prerequisites

• Python 3.9+ installed
• Libraries: FastAPI, uvicorn, requests, pydantic, sqlite3 (builtin)
• Basic knowledge of HTTP, JSON and command line

Step 1: Understand the scenario and choose the flow

We will create a simple Webhook receiver with FastAPI that validates the payload, checks a secret (HMAC) and updates a SQLite table. The typical flow: the sender posts JSON; the receiver validates and responds 200; on success it updates local data.

Step 2: Structure the project and install dependencies

Create a folder for the project and install the required libraries.

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

Step 3: Create the minimal SQLite database

We create a simple table to store received records. This makes it easy to verify that synchronization is working.

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

Step 4: Implement the Webhook receiver with HMAC validation

We use HMAC-SHA256 to ensure the request comes from a known source. The sender and the receiver share a secret. The sender includes an X-Signature header with the HMAC of the 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'}

Step 5: Test the Webhook locally (simulated sender)

We simulate the sender with requests. We generate the HMAC of the body and post to the local receiver at 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)

Step 6: Run the receiver

Start the FastAPI server with uvicorn and then run send_test.py to send the event.

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

Verify the result

Confirm you receive 200 OK in the sender console. Open the SQLite database and check the items table: the record with id 'item-123' should exist with data_json containing the attributes.

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

Conclusion

You now have a functional Webhook receiver that validates HMAC, parses JSON with Pydantic and updates a local database. Next steps: store the SECRET in environment variables, add logging, handle retries and process events in the background (queue). Tip: when testing public exposure, try using ngrok to receive external Webhooks securely.