← Back to all products
$19
Webhook Trigger
Event-driven webhook dispatcher: map events to endpoints, render templated payloads, sign with HMAC-SHA256, POST via urllib, retry with backoff, and dead-letter failed deliveries.
JSONMarkdownPython
📄 Product Preview
Try the interactive reader and demo tools below, or get the full product with all content unlocked.
📖 Interactive Reader (Free Preview) ⚙ Try Demo Tools 📦 Download Free Sample📁 File Structure 10 files
webhook-trigger/
├── LICENSE
├── README.md
├── examples/
│ └── webhook_trigger_config.json
├── free-sample.zip
├── guide/
│ ├── 01_features.md
│ ├── 02_quick-start.md
│ ├── 03_output.md
│ └── 04_license.md
├── index.html
└── src/
└── webhook_trigger.py
📖 Documentation Preview README excerpt
Webhook Trigger
An event-driven webhook dispatcher built on Python stdlib. Register triggers mapping event names to one or more webhook endpoints. On emit, render a payload from a configurable template, sign it with HMAC-SHA256, POST it via urllib, and retry with exponential backoff on failure. Permanently failed deliveries go to a dead-letter log.
Features
- Event-to-endpoint mapping — Register triggers that fan out a single event to multiple webhook URLs
- Payload templating — Render payloads from
{{key}}templates substituted with event data - HMAC-SHA256 signing — Every outbound request carries a verifiable
X-Webhook-Signatureheader - Retry with exponential backoff — Configurable max retries and base delay for transient failures
- Dead-letter log — Permanently failed deliveries are persisted to a JSON file for later inspection
- Per-endpoint rate limiting — Token-bucket rate limiter prevents overwhelming downstream services
- Delivery log — Full audit trail with status, attempt count, timestamps, and error details
- Dry-run mode —
--testrenders and signs payloads without sending, safe to run offline
Requirements
- Python 3.10+
- No external dependencies (stdlib only)
Quick Start
# List all registered triggers
python src/webhook_trigger.py --config examples/webhook_trigger_config.json --list
# Emit an event (live delivery)
python src/webhook_trigger.py --config examples/webhook_trigger_config.json \
--emit order.created --data '{"id": 42, "amount": "99.95", "currency": "USD", "sku": "WIDGET-7", "quantity": "3"}'
# Dry-run — render + sign but do NOT send (works offline)
python src/webhook_trigger.py --config examples/webhook_trigger_config.json \
--emit order.created --data '{"id": 42, "amount": "99.95", "currency": "USD"}' --test
# Verbose output
python src/webhook_trigger.py --config examples/webhook_trigger_config.json \
--emit user.signup --data '{"id": 7, "email": "user@example.com", "plan": "pro"}' --test --log-level DEBUG
Configuration Reference
CLI Flags
| Flag | Default | Description |
|---|---|---|
--config, -c | (required) | Path to the trigger config file (JSON) |
--emit, -e | — | Event name to emit (e.g. order.created) |
--data, -d | {} | JSON payload data for the event |
--test, --dry-run | off | Render and sign but do not send |
--list, -l | off | List all registered triggers and exit |
--log-level | INFO | DEBUG, INFO, WARNING, or ERROR |
Config File Schema
{
"secret": "whsec_your_signing_secret",
"max_retries": 3,
"retry_base_delay": 1.0,
"rate_limit_per_second": 10,
*... continues with setup instructions, usage examples, and more.*
📄 Code Sample .py preview
src/webhook_trigger.py
#!/usr/bin/env python3
"""
Webhook Trigger — Automation Hub (DataNest)
An event-driven webhook dispatcher. Register triggers mapping event names to
one or more webhook endpoints. On emit, render a payload from a template
(substituting event data), sign it with HMAC-SHA256, POST it via urllib, and
on failure retry with exponential backoff. Permanently failed deliveries go
to a dead-letter log.
Usage:
python webhook_trigger.py --config webhook_config.json --emit order.created --data '{"id": 42}'
python webhook_trigger.py --config webhook_config.json --emit order.created --data '{"id": 42}' --test
python webhook_trigger.py --config webhook_config.json --list
Dependencies: Python 3.10+ stdlib only (no pip packages)
License: MIT
"""
from __future__ import annotations
import argparse
import hashlib
import hmac
import json
import logging
import re
import threading
import time
import urllib.error
import urllib.request
from dataclasses import dataclass, field
from datetime import datetime, timezone
from pathlib import Path
from typing import Any
# ---------------------------------------------------------------------------
# Constants
# ---------------------------------------------------------------------------
LOG_FORMAT = "%(asctime)s [%(levelname)s] %(name)s: %(message)s"
DEFAULT_MAX_RETRIES = 3
DEFAULT_RETRY_BASE_DELAY = 1.0
DEFAULT_RATE_LIMIT = 10.0
DEFAULT_HTTP_TIMEOUT = 30
SIGNATURE_HEADER = "X-Webhook-Signature"
TIMESTAMP_HEADER = "X-Webhook-Timestamp"
logger = logging.getLogger("webhook_trigger")
# ... 624 more lines ...