import argparse
import asyncio
import json
import logging
import os
import sqlite3
import time
from typing import Any, Dict, List, Optional
import httpx
from fastapi import FastAPI, Form, HTTPException, Request
from fastapi.responses import HTMLResponse
from fastapi.staticfiles import StaticFiles
from fastapi.templating import Jinja2Templates
# Configure logging
logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s")
app = FastAPI()
# Database setup
DATABASE_URL = "webhooks.db"
def create_table() -> None:
"""Creates the webhooks table in the SQLite database if it doesn't exist."""
conn = sqlite3.connect(DATABASE_URL)
cursor = conn.cursor()
cursor.execute(
"""
CREATE TABLE IF NOT EXISTS webhooks (
id INTEGER PRIMARY KEY AUTOINCREMENT,
timestamp TEXT DEFAULT CURRENT_TIMESTAMP,
payload TEXT
)
"""
)
conn.commit()
conn.close()
def save_webhook(payload: str) -> None:
"""Saves the webhook payload to the SQLite database."""
conn = sqlite3.connect(DATABASE_URL)
cursor = conn.cursor()
cursor.execute("INSERT INTO webhooks (payload) VALUES (?)", (payload,))
conn.commit()
conn.close()
def get_recent_webhooks(limit: int = 10) -> List[Dict[str, Any]]:
"""Retrieves the most recent webhooks from the database."""
conn = sqlite3.connect(DATABASE_URL)
cursor = conn.cursor()
cursor.execute("SELECT timestamp, payload FROM webhooks ORDER BY timestamp DESC LIMIT ?", (limit,))
rows = cursor.fetchall()
conn.close()
return [{"timestamp": row[0], "payload": row[1]} for row in rows]
# FastAPI endpoints
@app.post("/webhook")
async def receive_webhook(request: Request) -> None:
"""Receives a webhook and saves the payload to the database."""
try:
payload = await request.body()
payload_str = payload.decode("utf-8")
save_webhook(payload_str)
logging.info("Webhook received and saved.")
return {"status": "Webhook received"}
except Exception as e:
logging.error(f"Error processing webhook: {e}")
raise HTTPException(status_code=500, detail=str(e))
@app.get("/")
async def index() -> HTMLResponse:
"""Serves the main page with a list of recent webhooks."""
webhooks = get_recent_webhooks()
return templates.TemplateResponse("index.html", {"webhooks": webhooks})
@app.post("/test_webhook")
async def test_webhook(payload: str = Form(...)) -> str:
"""Endpoint to manually send a test webhook."""
try:
save_webhook(payload)
return "Test webhook saved successfully."
except Exception as e:
logging.error(f"Error saving test webhook: {e}")
raise HTTPException(status_code=500, detail=str(e))
# Static files for UI
app.mount("/static", StaticFiles(directory="static"), name="static")
templates = Jinja2Templates(directory="templates")
# Argument parsing
def parse_args() -> argparse.Namespace:
"""Parses command line arguments."""
parser = argparse.ArgumentParser(description="Webhook Tester - A simple app to receive and inspect webhooks.")
parser.add_argument("--host", type=str, default="0.0.0.0", help="The host to bind to (default: 0.0.0.0)")
parser.add_argument("--port", type=int, default=8000, help="The port to listen on (default: 8000)")
return parser.parse_args()
if __name__ == "__main__":
args = parse_args()
# Create the database table if it doesn't exist
create_table()
# Create static files and templates directories if they don't exist
os.makedirs("static", exist_ok=True)
os.makedirs("templates", exist_ok=True)
# Create a basic index.html file if it doesn't exist
if not os.path.exists("templates/index.html"):
with open("templates/index.html", "w") as f:
f.write("""
Webhook Tester
Recent Webhooks
{% for webhook in webhooks %}
- {{ webhook.timestamp }}
{{ webhook.payload }}
{% endfor %}
""")
# Run the FastAPI app
import uvicorn
uvicorn.run(app, host=args.host, port=args.port)