Webhook with FastAPI — Complete Guide
In this tutorial, you will learn about Webhook with FastAPI. We cover key concepts, practical examples, and best practices to help you master this topic.
Learn to handle webhooks in FastAPI: signature verification dependency, async processing, background tasks, raw body access, provider-specific endpoints, and best practices for FastAPI webhook consumers.
What You Learn
You will learn how to implement webhook endpoints in FastAPI using its dependency injection system, access raw request bodies for HMAC verification, process webhooks asynchronously with background tasks, and structure FastAPI webhook code for type safety and testability.
Why It Matters
FastAPI is the leading Python async web framework. Its dependency injection system makes webhook verification clean and testable. Background tasks enable async processing without external task queues. Type validation ensures webhook payloads match expected schemas.
Real-World Use
DodaTech's FastAPI-based analytics service consumes webhooks from 4 providers. Dependency injection handles signature verification. Background tasks process events without blocking. Pydantic models validate payloads. The system processes 100K webhooks daily with 20ms average verification time.
Raw Body Access
from fastapi import FastAPI, Request, HTTPException, BackgroundTasks, Depends
from fastapi.responses import JSONResponse
import hmac
import hashlib
import json
from typing import Any, Dict
app = FastAPI()
# Dependency to get raw body
async def get_raw_body(request: Request) -> bytes:
body = await request.body()
return body
# Dependency to get parsed body
async def get_parsed_body(raw_body: bytes = Depends(get_raw_body)) -> Dict[str, Any]:
try:
return json.loads(raw_body)
except json.JSONDecodeError:
raise HTTPException(status_code=400, detail='Invalid JSON')
Expected output: Dependencies provide raw bytes for signature verification and parsed JSON for processing. Separation ensures verification uses untampered raw data.
Signature Verification Dependency
from fastapi import Header, HTTPException, Depends
import hmac
import hashlib
from typing import Optional
# Signature verification dependency
async def verify_webhook_signature(
request: Request,
x_webhook_signature: Optional[str] = Header(None),
x_webhook_timestamp: Optional[str] = Header(None),
):
if not x_webhook_signature:
raise HTTPException(status_code=401, detail='Missing signature')
raw_body = await request.body()
secret = b'whsec_your_secret_here'
expected_sig = x_webhook_signature.replace('sha256=', '')
computed_sig = hmac.new(
secret,
raw_body,
hashlib.sha256
).hexdigest()
if not hmac.compare_digest(computed_sig, expected_sig):
raise HTTPException(status_code=401, detail='Invalid signature')
# Check timestamp for replay prevention
if x_webhook_timestamp:
from datetime import datetime, timezone
try:
webhook_time = datetime.fromisoformat(x_webhook_timestamp.replace('Z', '+00:00'))
age = (datetime.now(timezone.utc) - webhook_time).total_seconds()
if abs(age) > 300: # 5 minutes
raise HTTPException(status_code=400, detail='Stale webhook')
except ValueError:
raise HTTPException(status_code=400, detail='Invalid timestamp')
return True
Expected output: FastAPI dependency extracts headers using type annotations, reads raw body, verifies HMAC signature with timing-safe comparison, validates timestamp, and raises HTTPException on failure. FastAPI converts these to proper HTTP error responses.
Pydantic Models
from pydantic import BaseModel, Field
from typing import Optional, List
from datetime import datetime
class PaymentData(BaseModel):
id: str
amount: int
currency: str = 'usd'
customer: str
status: str
created: datetime
class PaymentWebhook(BaseModel):
id: str
type: str
data: PaymentData
created: datetime
class GitHubPushPayload(BaseModel):
ref: str
before: str
after: str
repository: Dict[str, Any]
sender: Dict[str, Any]
commits: List[Dict[str, Any]]
class GitHubWebhook(BaseModel):
event: str # From header
payload: GitHubPushPayload
Expected output: Pydantic models validate webhook payloads. Type mismatches and missing fields are caught early and return 422 Unprocessable Entity with detailed error messages.
Webhook Endpoints
from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException
from typing import Dict, Any
webhook_router = APIRouter(prefix='/webhooks')
# In-memory event store
event_handlers: Dict[str, callable] = {}
def register_handler(event_type: str):
"""Decorator to register event handlers."""
def decorator(func):
event_handlers[event_type] = func
return func
return decorator
@webhook_router.post('/stripe')
async def stripe_webhook(
background_tasks: BackgroundTasks,
verified: bool = Depends(verify_webhook_signature),
body: Dict[str, Any] = Depends(get_parsed_body),
):
event_type = body.get('type', '')
handler = event_handlers.get(event_type)
if handler:
background_tasks.add_task(handler, body.get('data', {}))
return {'status': 'accepted', 'event': event_type}
return {'status': 'unhandled', 'event': event_type}
@webhook_router.post('/github')
async def github_webhook(
background_tasks: BackgroundTasks,
x_github_event: str = Header(...),
verified: bool = Depends(verify_webhook_signature),
body: Dict[str, Any] = Depends(get_parsed_body),
):
handler = event_handlers.get(x_github_event)
if handler:
background_tasks.add_task(handler, body)
return {'status': 'accepted', 'event': x_github_event}
return {'status': 'unhandled', 'event': x_github_event}
@register_handler('payment_intent.succeeded')
async def handle_payment_succeeded(data: Dict[str, Any]):
print(f"Payment succeeded: {data.get('id')}")
# Process payment logic here
await update_database(data)
@register_handler('push')
async def handle_github_push(data: Dict[str, Any]):
repo = data.get('repository', {}).get('full_name', 'unknown')
print(f"Push to {repo}")
# Process push event here
Expected output: Endpoints register with prefix, verify signatures via dependency injection, dispatch to registered handlers via background tasks, and return immediately. Handlers are registered with a decorator for clean organization.
Background Task Processing
from fastapi import BackgroundTasks
import asyncio
from typing import Dict, Any
async def process_webhook_event(event_type: str, data: Dict[str, Any]):
"""Async background task with retry logic."""
max_retries = 3
for attempt in range(max_retries):
try:
await execute_handler(event_type, data)
log_webhook_success(event_type, data)
return
except Exception as e:
log_webhook_failure(event_type, data, str(e), attempt)
if attempt < max_retries - 1:
await asyncio.sleep(2 ** attempt * 60)
else:
log_webhook_dead_letter(event_type, data, str(e))
async def execute_handler(event_type: str, data: Dict[str, Any]):
handler = event_handlers.get(event_type)
if handler:
if asyncio.iscoroutinefunction(handler):
await handler(data)
else:
handler(data)
Expected output: Background tasks run asynchronously after the response is sent. Failed tasks retry with exponential backoff. Permanent failures are logged to dead letter queue.
Provider-Specific Verification
# Stripe-specific verification
async def verify_stripe_signature(
request: Request,
stripe_signature: Optional[str] = Header(None),
):
import stripe
from django.conf import settings
if not stripe_signature:
raise HTTPException(status_code=401, detail='Missing Stripe signature')
raw_body = await request.body()
try:
event = stripe.Webhook.construct_event(
raw_body,
stripe_signature,
settings.STRIPE_WEBHOOK_SECRET
)
return event
except ValueError:
raise HTTPException(status_code=400, detail='Invalid payload')
except stripe.error.SignatureVerificationError:
raise HTTPException(status_code=401, detail='Invalid signature')
@webhook_router.post('/stripe-official')
async def stripe_webhook_official(
background_tasks: BackgroundTasks,
event: stripe.Event = Depends(verify_stripe_signature),
):
background_tasks.add_task(handle_stripe_event, event)
return {'status': 'accepted'}
Expected output: Provider-specific dependencies encapsulate signature verification. Stripe uses its official library. Custom providers use HMAC. The endpoint only receives verified event objects.
Common Mistakes
1. Not Using BackgroundTasks
Processing in the request handler blocks the server. FastAPI runs sync handlers in a thread pool but they still block. Use BackgroundTasks for webhook processing. The response returns immediately.
2. Missing Request Body for Dependencies
FastAPI dependencies cannot read request.body() twice by default. Use a single dependency that returns both raw and parsed body. Or configure FastAPI to allow body re-reading.
3. Not Using Pydantic for Validation
Dict[str, Any] does not validate payload structure. Define Pydantic models for each event type. FastAPI automatically returns 422 on validation failure with detailed error messages.
4. Sharing Secret Across Environments
Development, staging, and production should use different webhook secrets. Use environment variables or settings. Never hardcode secrets. Use a dependency that reads the secret from configuration.
5. No Rate Limiting
FastAPI webhook endpoints can receive high traffic. Use slowapi or custom middleware for rate limiting. Set different limits per provider endpoint. Return 429 when limits are exceeded.
Practice Questions
1. How does FastAPI dependency injection help with webhooks?
Dependencies encapsulate verification logic. They extract headers, read bodies, verify signatures, and validate timestamps. The endpoint only receives verified data. Dependencies are testable and reusable.
2. Why use BackgroundTasks instead of asyncio.create_task?
BackgroundTasks are designed for FastAPI. They run after the response is sent, preventing client timeouts. FastAPI manages the task lifecycle. asyncio.create_task may run before response completion.
3. How do you access the raw request body in FastAPI?
Use await request.body() inside a dependency. The body is consumed once. Return both raw bytes and parsed dict from the same dependency to avoid double-reading.
4. How do you handle different providers with different verification in FastAPI?
Create separate dependencies for each provider. Use the dependency in the provider-specific route. Provider-specific logic is isolated in its own dependency function. Routes are clean and focused.
Challenge
Build a FastAPI webhook ingestion service: endpoints for Stripe, GitHub, and SendGrid, provider-specific signature verification dependencies, Pydantic models for each provider's payload, background task processing with retry, rate limiting with slowapi, and Prometheus metrics for webhook throughput.
FAQ
Mini Project: FastAPI Webhook Service
Build a FastAPI webhook service that: accepts POST at /webhooks/{provider}, verifies provider-specific signatures via dependencies, validates payloads with Pydantic models, processes events with BackgroundTasks and retry logic, exposes /webhooks/log for delivery history, and provides /webhooks/stats for throughput metrics.
What's Next
Now that you can handle webhooks in FastAPI, learn about Database Storage for Webhooks to persist webhook events reliably.
Built by the developers of DodaTech
Doda Browser, DodaZIP & Durga Antivirus Pro