README.md

Webhook Processor Demo

A simple FastAPI-based webhook receiver that validates incoming webhook payloads and simulates notification processing.

What This Demo Does

This project simulates a webhook receiver endpoint that:

• Accepts POST requests with JSON payloads containing id, event, and metadata fields
• Normalizes all keys to lowercase for consistent processing
• Validates required fields and returns appropriate error codes
• Logs received events to the console
• Calls a mock notify() function that simulates sending notifications

The notify() function is a placeholder that would typically integrate with real notification services like Slack, Telegram, Email, or other automation systems in a production environment.

Requirements

• Python 3.7+
• fastapi
• uvicorn

Install dependencies:

pip install fastapi uvicorn

Running the Server

Start the server with:

uvicorn webhook_server:app --reload

The server will start on http://localhost:8000 by default.

Testing the Webhook

Using curl

curl -X POST http://localhost:8000/webhook \
  -H "Content-Type: application/json" \
  -d @test_payload.json

Using HTTPie

http POST http://localhost:8000/webhook < test_payload.json

Expected Response

{
  "status": "success",
  "message": "Webhook processed for event: order.created",
  "event_id": "evt_123"
}

Expected Console Output

2024-XX-XX XX:XX:XX - INFO - Received webhook event: order.created (id: evt_123)
2024-XX-XX XX:XX:XX - INFO - Metadata: {'user': 'demo_user', 'amount': 99}
Notification sent for event: order.created

Error Handling

• 400 Bad Request: Returned when required fields (id, event, metadata) are missing
• 422 Unprocessable Entity: Returned when the request body is not valid JSON
• 500 Internal Server Error: Returned for unexpected server errors

Project Structure

webhook-processor-demo/
├── webhook_server.py    # FastAPI application
├── test_payload.json    # Example webhook payload
└── README.md            # This file

test_payload.json

{
  "id": "evt_123",
  "event": "order.created",
  "metadata": {
    "user": "demo_user",
    "amount": 99
  }
}

webhook_server.py

"""
Simple FastAPI webhook receiver demo.
Validates incoming webhook payloads and simulates notifications.
"""

from fastapi import FastAPI, Request, HTTPException
from fastapi.responses import JSONResponse
import json
import logging

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)

app = FastAPI(title="Webhook Processor Demo")


def notify(event_name: str) -> None:
    """
    Mock notification function.
    In production, this would send notifications via Slack, Telegram, Email, etc.
    """
    print(f"Notification sent for event: {event_name}")


def normalize_keys(data: dict) -> dict:
    """
    Recursively normalize all keys in a dictionary to lowercase.
    """
    if isinstance(data, dict):
        return {k.lower(): normalize_keys(v) for k, v in data.items()}
    elif isinstance(data, list):
        return [normalize_keys(item) for item in data]
    else:
        return data


@app.post("/webhook")
async def webhook_handler(request: Request):
    """
    Handle incoming webhook POST requests.
    
    Expected JSON payload:
    {
        "id": "evt_123",
        "event": "order.created",
        "metadata": {...}
    }
    """
    try:
        # Parse JSON body
        try:
            body = await request.json()
        except json.JSONDecodeError:
            raise HTTPException(
                status_code=422,
                detail="Invalid JSON format"
            )
        
        # Normalize keys to lowercase
        normalized_body = normalize_keys(body)
        
        # Validate required fields
        required_fields = ["id", "event", "metadata"]
        missing_fields = [field for field in required_fields if field not in normalized_body]
        
        if missing_fields:
            raise HTTPException(
                status_code=400,
                detail=f"Missing required fields: {', '.join(missing_fields)}"
            )
        
        # Extract validated data
        event_id = normalized_body["id"]
        event_name = normalized_body["event"]
        metadata = normalized_body["metadata"]
        
        # Log the received event
        logger.info(f"Received webhook event: {event_name} (id: {event_id})")
        logger.info(f"Metadata: {metadata}")
        
        # Send notification
        notify(event_name)
        
        return JSONResponse(
            status_code=200,
            content={
                "status": "success",
                "message": f"Webhook processed for event: {event_name}",
                "event_id": event_id
            }
        )
    
    except HTTPException:
        # Re-raise HTTP exceptions (400, 422)
        raise
    except Exception as e:
        # Handle any unexpected errors
        logger.error(f"Unexpected error processing webhook: {str(e)}")
        raise HTTPException(
            status_code=500,
            detail="Internal server error"
        )


@app.get("/")
async def root():
    """Health check endpoint."""
    return {"status": "ok", "message": "Webhook processor is running"}


if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000, reload=True)