FastAPI自学教程（74） - OpenAPI Webhooks

1.基础用法

from datetime import datetime
from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class Subscription(BaseModel):
    username: str
    monthly_fee: float
    start_date: datetime

@app.webhooks.post("new-subscription")
def new_subscription(body: Subscription):
    """
    当有新用户订阅服务时，将POST请求发送到用户注册的URL
    请求体包含用户信息、订阅费用和开始时间
    """

• 通过@app.webhooks.post()装饰器定义Webhook事件，参数为事件标识符（如”new-subscription”）
• Webhook请求体使用Pydantic模型验证，支持自动生成OpenAPI文档
• 处理流程：

  graph TD
    A[用户触发事件] --> B[FastAPI构造请求体]
    B --> C[发送至用户注册的URL]
    C --> D[外部API处理请求]
    D --> E[返回响应状态]

2.Webhook请求体验证

class WebhookPayload(BaseModel):
    event: str
    data: dict

@app.webhooks.post("payment-success")
def payment_notify(body: WebhookPayload):
    """支付成功通知"""

• 请求体字段自动验证机制：
  • 类型校验（如event必须为字符串）
  • 数据格式转换（日期时间自动转换为datetime对象）
  • 错误自动生成（返回422状态码和错误详情）

3.生产环境建议

1. 安全验证：

   • 使用签名验证（如Svix库）防止伪造请求
   • 校验请求时间戳防止重放攻击（建议±5分钟容差）

2. URL注册机制：

   • 通过管理后台让用户配置接收URL
   • 支持HTTPS协议和URL有效性校验

3. 错误重试策略：

   from tenacity import retry, stop_after_attempt
   
   @retry(stop=stop_after_attempt(3))
   async def send_webhook(url: str, data: dict):
       async with httpx.AsyncClient() as client:
           return await client.post(url, json=data)

   • 建议指数退避重试机制

4. 异步处理：

   • 使用后台任务避免阻塞主线程

     from fastapi import BackgroundTasks
     
     @app.post("/payments/")
     async def create_payment(background: BackgroundTasks):
         background.add_task(send_webhook, url, data)

安全验证示例参考：Svix Webhook验证实现
测试命令示例：

curl -X POST http://localhost:8000/payments/ \
-H "Content-Type: application/json" \
-d '{"amount": 99.99, "user_id": 123}'