Webhook Triggers

Provides patterns for sending and receiving webhooks in invoice workflows.

Sending Webhooks

import requests
import hmac
import hashlib
import json
from typing import Dict, Any, Optional

def send_webhook(
    url: str,
    payload: Dict[str, Any],
    headers: Dict[str, str] = None,
    timeout: int = 30
) -> tuple[bool, int, str]:
    """
    Send webhook POST request.

    Returns:
        Tuple of (success, status_code, response_text)

    Usage:
        success, status, response = send_webhook(
            url="https://api.example.com/webhooks/invoice",
            payload={"event": "invoice.created", "data": {...}}
        )
    """
    headers = headers or {}
    headers.setdefault("Content-Type", "application/json")

    try:
        response = requests.post(
            url,
            json=payload,
            headers=headers,
            timeout=timeout
        )
        return response.ok, response.status_code, response.text
    except requests.RequestException as e:
        return False, 0, str(e)

Signed Webhooks

def create_webhook_signature(payload: Dict, secret: str) -> str:
    """
    Create HMAC signature for webhook payload.

    Usage:
        signature = create_webhook_signature(payload, "webhook_secret")
        headers["X-Webhook-Signature"] = signature
    """
    payload_bytes = json.dumps(payload, sort_keys=True).encode()
    signature = hmac.new(
        secret.encode(),
        payload_bytes,
        hashlib.sha256
    ).hexdigest()
    return f"sha256={signature}"

def send_signed_webhook(
    url: str,
    payload: Dict[str, Any],
    secret: str,
    headers: Dict[str, str] = None
) -> tuple[bool, int, str]:
    """
    Send webhook with HMAC signature.

    Usage:
        success, status, response = send_signed_webhook(
            url="https://api.example.com/webhooks",
            payload={"event": "invoice.paid", "invoice_id": "INV-001"},
            secret="webhook_secret_key"
        )
    """
    headers = headers or {}
    headers["X-Webhook-Signature"] = create_webhook_signature(payload, secret)
    headers["X-Webhook-Timestamp"] = str(int(time.time()))

    return send_webhook(url, payload, headers)

def verify_webhook_signature(payload: Dict, signature: str, secret: str) -> bool:
    """
    Verify incoming webhook signature.

    Usage:
        if verify_webhook_signature(payload, request.headers["X-Webhook-Signature"], secret):
            process_webhook(payload)
    """
    expected = create_webhook_signature(payload, secret)
    return hmac.compare_digest(signature, expected)

Invoice Event Webhooks

import time
from datetime import datetime

def create_invoice_event_payload(
    event_type: str,
    invoice_data: Dict[str, Any]
) -> Dict[str, Any]:
    """
    Create standardized invoice event payload.

    Event types:
    - invoice.created
    - invoice.updated
    - invoice.approved
    - invoice.paid
    - invoice.overdue

    Usage:
        payload = create_invoice_event_payload(
            event_type="invoice.created",
            invoice_data={
                "invoice_number": "INV-001",
                "amount": 1500.00,
                "vendor": "Acme Corp"
            }
        )
    """
    return {
        "event": event_type,
        "timestamp": datetime.utcnow().isoformat() + "Z",
        "data": invoice_data
    }

def trigger_invoice_webhook(
    webhook_url: str,
    event_type: str,
    invoice_data: Dict[str, Any],
    secret: str = None
):
    """
    Trigger webhook for invoice event.

    Usage:
        trigger_invoice_webhook(
            webhook_url="https://api.erp.com/webhooks",
            event_type="invoice.approved",
            invoice_data=invoice,
            secret="webhook_secret"
        )
    """
    payload = create_invoice_event_payload(event_type, invoice_data)

    if secret:
        return send_signed_webhook(webhook_url, payload, secret)
    return send_webhook(webhook_url, payload)

Webhook with Retry

import time

def send_webhook_with_retry(
    url: str,
    payload: Dict[str, Any],
    max_retries: int = 3,
    retry_delay: float = 1.0,
    backoff_factor: float = 2.0
) -> tuple[bool, int, str]:
    """
    Send webhook with exponential backoff retry.

    Usage:
        success, status, response = send_webhook_with_retry(
            url="https://api.example.com/webhooks",
            payload={"event": "invoice.created"},
            max_retries=5
        )
    """
    delay = retry_delay

    for attempt in range(max_retries + 1):
        success, status, response = send_webhook(url, payload)

        if success:
            return success, status, response

        if attempt < max_retries:
            time.sleep(delay)
            delay *= backoff_factor

    return False, status, response

Batch Webhook Triggers

from typing import List

def trigger_batch_webhooks(
    webhook_url: str,
    events: List[Dict[str, Any]],
    batch_size: int = 10
) -> tuple[int, int]:
    """
    Trigger webhooks for batch of events.

    Returns:
        Tuple of (successful_count, failed_count)
    """
    successful = 0
    failed = 0

    for event in events:
        success, _, _ = send_webhook(webhook_url, event)
        if success:
            successful += 1
        else:
            failed += 1

    return successful, failed

def trigger_multi_endpoint_webhook(
    urls: List[str],
    payload: Dict[str, Any]
) -> Dict[str, bool]:
    """
    Send same webhook to multiple endpoints.

    Usage:
        results = trigger_multi_endpoint_webhook(
            urls=["https://api1.com/webhook", "https://api2.com/webhook"],
            payload={"event": "invoice.paid"}
        )
    """
    results = {}
    for url in urls:
        success, _, _ = send_webhook(url, payload)
        results[url] = success
    return results

Helper Script

Use helper.py for the WebhookClient class with comprehensive retry and signing support.