Webhook Handling: Stripe and GitHub

Webhooks are HTTP POST requests that third-party services send to your app when events happen: a payment succeeds, a pull request is opened, a subscription is cancelled. Here's how to handle them correctly.

The Golden Rules of Webhook Handling

1. Respond with 200 immediately — the sender will retry if you take too long
2. Verify the signature — confirm the request is genuinely from the sender
3. Be idempotent — the same event may be delivered more than once; handle it gracefully
4. Do real work in the background — queue heavy processing via Celery

Database Table for Idempotency

CREATE TABLE webhook_events (
    id SERIAL PRIMARY KEY,
    source VARCHAR(50) NOT NULL,       -- 'stripe', 'github'
    event_id VARCHAR(255) NOT NULL,    -- External event ID
    event_type VARCHAR(100) NOT NULL,
    payload JSONB NOT NULL,
    processed BOOLEAN DEFAULT FALSE,
    received_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    UNIQUE(source, event_id)           -- Prevents duplicate processing
);

Stripe Webhooks

Setup

In the Stripe Dashboard: Developers → Webhooks → Add endpoint → your URL.

STRIPE_WEBHOOK_SECRET=whsec_your_secret_here

The Endpoint

import stripe
import os
from flask import request, jsonify
from utils.db import get_db

stripe.api_key = os.environ.get("STRIPE_SECRET_KEY")

@app.route("/webhooks/stripe", methods=["POST"])
def stripe_webhook():
    payload = request.get_data(as_text=True)
    sig_header = request.headers.get("Stripe-Signature")
    webhook_secret = os.environ.get("STRIPE_WEBHOOK_SECRET")

    # Step 1: Verify the signature
    try:
        event = stripe.Webhook.construct_event(
            payload, sig_header, webhook_secret
        )
    except stripe.SignatureVerificationError:
        return jsonify({"error": "Invalid signature"}), 400

    # Step 2: Store for idempotency, skip if already processed
    with get_db() as conn:
        with conn.cursor() as cur:
            try:
                cur.execute("""
                    INSERT INTO webhook_events (source, event_id, event_type, payload)
                    VALUES ('stripe', %s, %s, %s)
                """, (event["id"], event["type"], payload))
                conn.commit()
            except Exception:
                # UNIQUE constraint violated — already received this event
                conn.rollback()
                return jsonify({"status": "already processed"}), 200

    # Step 3: Respond immediately, process asynchronously
    handle_stripe_event.delay(event["id"], event["type"], event["data"])
    return jsonify({"status": "received"}), 200

Handling Stripe Event Types

# tasks/webhook_tasks.py
from app import celery
from utils.db import get_db

@celery.task
def handle_stripe_event(event_id: str, event_type: str, data: dict):
    """Process a Stripe event in the background."""

    if event_type == "checkout.session.completed":
        session = data["object"]
        customer_email = session.get("customer_email")
        amount = session.get("amount_total")

        with get_db() as conn:
            with conn.cursor() as cur:
                cur.execute("""
                    UPDATE orders
                    SET status = 'paid', paid_at = NOW()
                    WHERE stripe_session_id = %s
                """, (session["id"],))
                conn.commit()

    elif event_type == "customer.subscription.deleted":
        subscription = data["object"]
        # Revoke access, send cancellation email, etc.

    elif event_type == "invoice.payment_failed":
        invoice = data["object"]
        # Notify user, update subscription status

    # Mark as processed
    with get_db() as conn:
        with conn.cursor() as cur:
            cur.execute(
                "UPDATE webhook_events SET processed = TRUE WHERE source = 'stripe' AND event_id = %s",
                (event_id,)
            )
            conn.commit()

GitHub Webhooks

Setup

In your GitHub repo: Settings → Webhooks → Add webhook.

GITHUB_WEBHOOK_SECRET=your-secret-here

The Endpoint

import hmac

@app.route("/webhooks/github", methods=["POST"])
def github_webhook():
    # Step 1: Verify signature (HMAC-SHA256)
    secret = os.environ.get("GITHUB_WEBHOOK_SECRET", "").encode()
    signature = request.headers.get("X-Hub-Signature-256", "")
    payload = request.get_data()

    expected = "sha256=" + hmac.digest(secret, payload, "sha256").hex()

    if not hmac.compare_digest(expected, signature):
        return jsonify({"error": "Invalid signature"}), 400

    # Step 2: Parse the event
    event_type = request.headers.get("X-GitHub-Event")
    delivery_id = request.headers.get("X-GitHub-Delivery")
    data = request.get_json()

    # Step 3: Store for idempotency
    with get_db() as conn:
        with conn.cursor() as cur:
            try:
                cur.execute("""
                    INSERT INTO webhook_events (source, event_id, event_type, payload)
                    VALUES ('github', %s, %s, %s)
                """, (delivery_id, event_type, request.get_data(as_text=True)))
                conn.commit()
            except Exception:
                conn.rollback()
                return jsonify({"status": "already processed"}), 200

    # Step 4: Handle the event
    if event_type == "push":
        branch = data["ref"].replace("refs/heads/", "")
        commit_message = data["head_commit"]["message"]
        # Trigger a deployment, notify Slack, update a dashboard, etc.

    elif event_type == "pull_request":
        action = data["action"]  # opened, closed, merged
        pr_title = data["pull_request"]["title"]

    return jsonify({"status": "received"}), 200

Testing Webhooks Locally

Use the Stripe CLI to forward events to localhost:

stripe listen --forward-to localhost:5000/webhooks/stripe

For GitHub, use ngrok to expose your local server:

ngrok http 5000
# Use the ngrok URL as your webhook endpoint in GitHub settings

Viewing Webhook History

@app.route("/admin/webhooks")
@admin_required
def webhook_log():
    with get_db() as conn:
        with conn.cursor() as cur:
            cur.execute("""
                SELECT source, event_type, processed, received_at
                FROM webhook_events
                ORDER BY received_at DESC
                LIMIT 100
            """)
            events = cur.fetchall()
    return render_template("admin/webhooks.html", events=events)