Django 中的 Webhook：综合指南-Python教程-PHP中文网

Webhooks in Django: A Comprehensive Guide

Webhooks 是创建实时事件驱动应用程序的强大功能。在 Django 生态系统中，它们使应用程序能够近乎实时地对外部事件做出反应，这使得它们对于与第三方服务（例如支付网关、社交媒体平台或数据监控系统）的集成特别有用。本指南将介绍 Webhook 的基础知识、在 Django 中设置它们的过程，以及构建健壮、可扩展且安全的 Webhook 处理系统的最佳实践。

什么是 Webhook？

Webhooks 是 HTTP 回调，每当特定事件发生时，它就会将数据发送到外部 URL。与您的应用程序请求数据的传统 API 不同，Webhook 允许外部服务根据某些触发器将数据“推送”到您的应用程序。

例如，如果您的应用程序与支付处理器集成，则每次支付成功或失败时，Webhook 可能会通知您。事件数据（通常采用 JSON 格式）作为 POST 请求发送到应用程序中的指定端点，使其能够根据需要处理或存储信息。

为什么使用 Webhook？

Webhooks 提供了反应式和事件驱动的模型。他们的主要优点包括：

实时数据流：事件发生后立即接收更新。
减少轮询：无需不断检查更新。
简单集成：与 Stripe、GitHub 或 Slack 等第三方服务连接。
可扩展性：有效处理大量事件和响应

在 Django 中设置 Webhook

在 Django 中实现 Webhook 涉及创建专用视图来接收和处理传入的 POST 请求。让我们完成这些步骤。

第 1 步：设置 Webhook URL

创建专门用于处理 Webhook 请求的 URL 端点。例如，假设我们正在为支付服务设置一个 Webhook，该服务会在交易完成时通知我们。

在 urls.py 中：

from django.urls import path
from . import views

urlpatterns = [
    path("webhook/", views.payment_webhook, name="payment_webhook"),
]

登录后复制

第 2 步：创建 Webhook 视图

视图处理传入的请求并处理接收到的数据。由于 Webhooks 通常发送 JSON 有效负载，因此我们将首先解析 JSON 并根据有效负载的内容执行必要的操作。
在views.py中：

import json
from django.http import JsonResponse, HttpResponseBadRequest
from django.views.decorators.csrf import csrf_exempt

@csrf_exempt  # Exempt this view from CSRF protection
def payment_webhook(request):
    if request.method != "POST":
        return HttpResponseBadRequest("Invalid request method.")

    try:
        data = json.loads(request.body)
    except json.JSONDecodeError:
        return HttpResponseBadRequest("Invalid JSON payload.")

    # Perform different actions based on the event type
    event_type = data.get("event_type")

    if event_type == "payment_success":
        handle_payment_success(data)
    elif event_type == "payment_failure":
        handle_payment_failure(data)
    else:
        return HttpResponseBadRequest("Unhandled event type.")

    # Acknowledge receipt of the webhook
    return JsonResponse({"status": "success"})

登录后复制

第 3 步：实现辅助函数

为了保持视图的简洁和模块化，最好创建单独的函数来处理每个特定的事件类型。

from django.urls import path
from . import views

urlpatterns = [
    path("webhook/", views.payment_webhook, name="payment_webhook"),
]

登录后复制

步骤4：在第三方服务中配置Webhook

设置端点后，在您要集成的第三方服务中配置 Webhook URL。通常，您会在服务的仪表板中找到 Webhook 配置选项。第三方服务还可能提供选项来指定哪些事件应触发 Webhook。

Webhook 的安全最佳实践

由于 Webhooks 向外部数据开放您的应用程序，因此遵循安全最佳实践对于防止误用或数据泄露至关重要。

使用身份验证令牌：包含共享密钥或令牌来验证传入请求。许多服务在请求标头中提供了您可以验证的签名。

import json
from django.http import JsonResponse, HttpResponseBadRequest
from django.views.decorators.csrf import csrf_exempt

@csrf_exempt  # Exempt this view from CSRF protection
def payment_webhook(request):
    if request.method != "POST":
        return HttpResponseBadRequest("Invalid request method.")

    try:
        data = json.loads(request.body)
    except json.JSONDecodeError:
        return HttpResponseBadRequest("Invalid JSON payload.")

    # Perform different actions based on the event type
    event_type = data.get("event_type")

    if event_type == "payment_success":
        handle_payment_success(data)
    elif event_type == "payment_failure":
        handle_payment_failure(data)
    else:
        return HttpResponseBadRequest("Unhandled event type.")

    # Acknowledge receipt of the webhook
    return JsonResponse({"status": "success"})

登录后复制

速率限制 Webhooks：通过限制端点在一段时间内可以处理的请求数量来防止滥用。
快速响应：Webhooks 通常需要快速响应。避免在视图中进行复杂的同步处理以防止超时。
用于繁重处理的队列任务：如果 webhook 处理涉及长时间运行的任务，请使用像 Celery 这样的任务队列来异步处理它们。

def handle_payment_success(data):
    # Extract payment details and update your models or perform required actions
    transaction_id = data["transaction_id"]
    amount = data["amount"]
    # Logic to update the database or notify the user
    print(f"Payment succeeded with ID: {transaction_id} for amount: {amount}")

def handle_payment_failure(data):
    # Handle payment failure logic
    transaction_id = data["transaction_id"]
    reason = data["failure_reason"]
    # Logic to update the database or notify the user
    print(f"Payment failed with ID: {transaction_id}. Reason: {reason}")

登录后复制

测试网络钩子

测试 Webhooks 可能具有挑战性，因为它们需要外部服务来触发它们。以下是一些常见的测试方法：

使用像 ngrok 这样的服务：Ngrok 创建一个临时公共 URL，将请求转发到本地开发服务器，允许第三方服务向其发送 webhooks。
模拟请求：从测试脚本或Django的manage.py shell手动触发对webhook端点的请求。
Django 的测试客户端：通过模拟 POST 请求为 webhook 视图编写单元测试。

import hmac
import hashlib

def verify_signature(request):
    secret = "your_shared_secret"
    signature = request.headers.get("X-Signature")
    payload = request.body

    computed_signature = hmac.new(
        secret.encode(),
        payload,
        hashlib.sha256
    ).hexdigest()

    return hmac.compare_digest(computed_signature, signature)

登录后复制