Duplicate Webhooks: Build an Idempotent Inbox in Python

import sqlite3 import threading import time setup = sqlite3.connect("naive.db") setup.execute("CREATE TABLE IF NOT EXISTS seen (id TEXT)") setup.execute("DELETE FROM seen") setup.commit() work_done = [] def handle(delivery_id): db = sqlite3.connect("naive.db") if db.execute("SELECT 1 FROM seen WHERE id = ?", (delivery_id,)).fetchone(): return # already handled  time.sleep(0.2) # the real work takes a moment  work_done.append(delivery_id) db.execute("INSERT INTO seen VALUES (?)", (delivery_id,)) db.commit() copies = [threading.Thread(target=handle, args=("evt_1",)) for _ in range(2)] # two copies arrive together for copy in copies: copy.start() for copy in copies: copy.join() print(f"The work was done {len(work_done)} times") 

The work was done 2 times 

"""A durable inbox for webhook deliveries: store first, process later, and never twice.""" import os import secrets import sqlite3 import sys import time LEASE = 60 # seconds a worker may hold a delivery before another may take it over MAX_ATTEMPTS = 5 # after this many tries, a delivery is set aside for a person to look at RETRY_DELAY = 2 # seconds before the first retry, doubling each time  def connect(path=None): conn = sqlite3.connect(path or os.environ.get("INBOX_DB", "inbox.db"), timeout=30, isolation_level=None) # we run transactions ourselves  conn.execute("PRAGMA journal_mode=WAL") conn.execute("""CREATE TABLE IF NOT EXISTS deliveries ( id TEXT PRIMARY KEY, body TEXT, received REAL, status TEXT, attempts INTEGER DEFAULT 0, available_at REAL, leased_until REAL, owner TEXT, error TEXT)""") return conn def store(conn, delivery_id, body, now=time.time): """Save a delivery. Returns False if this ID has been seen before. The database enforces that, not our code.""" cursor = conn.execute("INSERT OR IGNORE INTO deliveries (id, body, received, status, available_at) VALUES (?, ?, ?, 'received', ?)", (delivery_id, body, now(), now())) return cursor.rowcount == 1 def claim(conn, now=time.time): """Take one delivery to work on, or return None. Two workers can never take the same one.""" conn.execute("BEGIN IMMEDIATE") # only one worker at a time gets past this line  try: t, owner = now(), secrets.token_hex(4) conn.execute("UPDATE deliveries SET status = 'failed', error = 'gave up: the worker kept crashing' " "WHERE status = 'processing' AND leased_until < ? AND attempts >= ?", (t, MAX_ATTEMPTS)) row = conn.execute("SELECT id, body FROM deliveries WHERE (status = 'received' AND available_at <= ?) " "OR (status = 'processing' AND leased_until < ?) ORDER BY received LIMIT 1", (t, t)).fetchone() # new work, or work whose worker vanished  if row: conn.execute("UPDATE deliveries SET status = 'processing', attempts = attempts + 1, leased_until = ?, owner = ? WHERE id = ?", (t + LEASE, owner, row[0])) conn.execute("COMMIT") return (row[0], row[1], owner) if row else None except Exception: conn.execute("ROLLBACK") raise def finish(conn, delivery_id, owner): """Mark a delivery done. Returns False if the lease was lost, meaning someone else took it over.""" return conn.execute("UPDATE deliveries SET status = 'done', leased_until = NULL, error = NULL " "WHERE id = ? AND owner = ? AND status = 'processing'", (delivery_id, owner)).rowcount == 1 def fail(conn, delivery_id, owner, error, now=time.time): """The work failed. Try again later, or set it aside after too many attempts.""" row = conn.execute("SELECT attempts FROM deliveries WHERE id = ? AND owner = ? AND status = 'processing'", (delivery_id, owner)).fetchone() if not row: return False if row[0] >= MAX_ATTEMPTS: conn.execute("UPDATE deliveries SET status = 'failed', error = ?, leased_until = NULL WHERE id = ?", (error, delivery_id)) else: conn.execute("UPDATE deliveries SET status = 'received', error = ?, leased_until = NULL, available_at = ? WHERE id = ?", (error, now() + RETRY_DELAY * 2 ** (row[0] - 1), delivery_id)) return True if __name__ == "__main__": conn = connect() if sys.argv[1:2] == ["status"]: for status, count in conn.execute("SELECT status, COUNT(*) FROM deliveries GROUP BY status ORDER BY status"): print(f"{status:<11}{count}") for delivery_id, attempts, error in conn.execute("SELECT id, attempts, error FROM deliveries WHERE status = 'failed'"): print(f" failed: {delivery_id} after {attempts} attempts: {error}") elif sys.argv[1:2] == ["retry"] and len(sys.argv) == 3: # put a failed delivery back in the queue, once you've fixed the cause  changed = conn.execute("UPDATE deliveries SET status = 'received', attempts = 0, error = NULL, available_at = ? WHERE id = ? AND status = 'failed'", (time.time(), sys.argv[2])).rowcount print("Queued again." if changed else "No failed delivery with that ID.") else: sys.exit("Usage: python3 inbox.py status | retry <delivery id>") 

import logging import sqlite3 from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from inbox import connect, store MAX_BODY = 1_000_000 class Receiver(BaseHTTPRequestHandler): def reply(self, code, text): self.send_response(code) self.send_header("Content-Length", str(len(text))) self.end_headers() self.wfile.write(text.encode()) def do_POST(self): # In real life, check the signature first. The delivery ID comes from the provider,  # for example GitHub's X-GitHub-Delivery header, or the event "id" inside a Stripe body.  delivery_id = self.headers.get("X-Delivery-Id") length = int(self.headers.get("Content-Length") or 0) if not delivery_id: return self.reply(400, "missing delivery id") if length > MAX_BODY: return self.reply(413, "too large") body = self.rfile.read(length).decode("utf-8", errors="replace") try: conn = connect() is_new = store(conn, delivery_id, body) conn.close() except sqlite3.Error as error: logging.error("Couldn't store %s: %s", delivery_id, error) return self.reply(503, "try again later") # not saved, so ask the provider to retry  logging.info("%s %s", "Stored" if is_new else "Ignored a duplicate of", delivery_id) self.reply(200, "stored" if is_new else "duplicate") # either way, the provider is told it's handled  def log_message(self, *args): pass if __name__ == "__main__": logging.basicConfig(level=logging.INFO, format="%(message)s") print("Listening on http://127.0.0.1:8000") ThreadingHTTPServer(("127.0.0.1", 8000), Receiver).serve_forever() 

import json import logging import sys import time from inbox import claim, connect, fail, finish log = logging.getLogger("worker") def record_payment(conn, delivery_id, event): """The example work: record a payment. The delivery ID is the table's key, so doing it twice changes nothing.""" conn.execute("CREATE TABLE IF NOT EXISTS payments (delivery_id TEXT PRIMARY KEY, customer TEXT, amount REAL)") conn.execute("INSERT OR IGNORE INTO payments VALUES (?, ?, ?)", (delivery_id, event["customer"], event["amount"])) def run(conn, handler, stop_when_empty=False): while True: job = claim(conn) if not job: if stop_when_empty: return time.sleep(1) continue delivery_id, body, owner = job try: handler(conn, delivery_id, json.loads(body)) except Exception as error: log.warning("%s failed: %s", delivery_id, error) fail(conn, delivery_id, owner, str(error)) else: if finish(conn, delivery_id, owner): log.info("%s done", delivery_id) else: log.warning("%s: the lease was lost while working, so someone else took over", delivery_id) if __name__ == "__main__": logging.basicConfig(level=logging.INFO, format="%(message)s") run(connect(), record_payment, stop_when_empty="--once" in sys.argv) 

python3 inbox_receiver.py 

import json import urllib.request def send(delivery_id, event): request = urllib.request.Request("http://127.0.0.1:8000", data=json.dumps(event).encode(), headers={"X-Delivery-Id": delivery_id}) with urllib.request.urlopen(request) as response: print(f"{delivery_id}: {response.read().decode()}") send("evt_100", {"customer": "Maria Lopez", "amount": 24.99}) send("evt_100", {"customer": "Maria Lopez", "amount": 24.99}) # the provider retried, so the same delivery arrives again send("evt_101", {"customer": "Sam Okafor", "amount": 12.50}) send("evt_100", {"customer": "Maria Lopez", "amount": 24.99}) # and again sendSource：https://dev.to/procwire/duplicate-webhooks-build-an-idempotent-inbox-in-python-1cmd
My favorites