Django 中的 Webhook：综合指南-小浪资源网

Django 中的 Webhook：综合指南

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

什么是 webhook？

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

例如，如果您的应用程序与支付处理器集成，则每次支付成功或失败时，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 步：实现辅助函数

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

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}")

登录后复制

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

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

webhook 的安全最佳实践

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

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

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)

登录后复制

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

# example of celery task usage from .tasks import process_payment_event  @csrf_exempt def payment_webhook(request):     if request.method == "post":         data = json.loads(request.body)         process_payment_event.delay(data)  # queue the task for async processing         return jsonresponse({"status": "accepted"})     return httpresponsebadrequest("invalid request method.")

登录后复制

测试网络钩子

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

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

from django.test import TestCase, Client import json  class WebhookTest(TestCase):     def setUp(self):         self.client = Client()      def test_payment_success_webhook(self):         payload = {             "event_type": "payment_success",             "transaction_id": "12345",             "amount": 100         }         response = self.client.post(             "/webhook/",             data=json.dumps(payload),             content_type="application/json"         )         self.assertEqual(response.status_code, 200)         self.assertEqual(response.json(), {"status": "success"})

登录后复制