webhook-sync-service

A small, production-shaped system-integration service in Python: it ingests events from external systems (via signed webhooks and a scheduled poll), authenticates outbound calls with OAuth2, normalizes the data, and persists it idempotently to SQL.

It's a sanitized reference implementation of patterns I run in production (today on n8n, AWS Lambda, and Cloudflare Workers) — rebuilt here as clean, readable Python with tests. Provider and domain names are generic; no real systems or credentials are involved.

What it demonstrates

• REST APIs — build (FastAPI endpoints) and consume (an OAuth2 client for an upstream provider)
• OAuth2 client-credentials — token fetch, caching, refresh-before-expiry (app/providers/crm_client.py)
• Webhooks — HMAC-SHA256 signature verification over the raw body, constant-time compare (app/security.py)
• Idempotent persistence — upsert on a (source, external_id) natural key, so replays/retries never duplicate (app/services/sync.py)
• Resilience — retries with exponential backoff on transient (5xx/network) provider failures
• SQL — SQLAlchemy 2.0 models, portable across SQLite (dev/test) and Postgres (prod)
• Tests & CI — pytest suite + GitHub Actions
• Deployment — Dockerfile + docker-compose (app + Postgres)

Architecture

                      ┌────────────────────────────┐
  POST /webhooks/{src}│  HMAC verify → normalize    │
  (real-time) ───────▶│            ↓                │
                      │      idempotent upsert ──────────▶  Postgres
  cron: poll provider │            ↑                │      (records + event_log)
  (safety net) ──────▶│  OAuth2 REST client (httpx) │
                      └────────────────────────────┘
                                   │
                      GET /records, /records/{id}  ◀── read API

Two triggers, one sync path: webhooks for real-time delivery and a poller as a safety net for missed events — the same dual-trigger design used in real integrations.

Quickstart

python -m venv .venv && source .venv/bin/activate   # Windows: .venv\Scripts\activate
pip install -r requirements.txt
cp .env.example .env        # set WEBHOOK_SECRET (defaults work for local SQLite)

uvicorn app.main:app --reload
# docs: http://localhost:8000/docs

Send a signed webhook

SECRET=$(grep WEBHOOK_SECRET .env | cut -d= -f2)
BODY='{"id":"ORD-1001","type":"order","status":"paid","amount":129.99}'
SIG=$(python -c "import hmac,hashlib,sys;print(hmac.new(sys.argv[1].encode(),sys.argv[2].encode(),hashlib.sha256).hexdigest())" "$SECRET" "$BODY")

curl -s localhost:8000/webhooks/orders -H "X-Signature: $SIG" -d "$BODY"
# {"success":true,"record_id":1,"created":true}

curl -s localhost:8000/records?source=orders

Run with Postgres

WEBHOOK_SECRET=dev-secret docker compose up --build

Poll the provider (cron trigger)

python -m app.poller          # */5 * * * *  python -m app.poller

Tests

pytest -q

Covers signature verification, idempotent upserts, multi-source isolation, and the webhook + read endpoints end to end.

Project layout

app/
  main.py            FastAPI app + lifespan
  config.py          env-driven settings (pydantic-settings)
  db.py              engine / session / Base
  models.py          Record + EventLog (natural-key uniqueness)
  schemas.py         API response models
  security.py        HMAC-SHA256 webhook verification
  providers/
    crm_client.py    OAuth2 client-credentials REST client (httpx) + retries
  routers/
    webhooks.py      POST /webhooks/{source}
    records.py       GET /records, /records/{id}
  services/
    sync.py          normalize + idempotent upsert
  poller.py          scheduled provider poll (cron)
tests/               pytest suite

Tech

Python 3.12 · FastAPI · SQLAlchemy 2.0 · httpx · Postgres · Docker · GitHub Actions

License

MIT — see LICENSE.