Django 中的 Webhook：综合指南

百变鹏仔 4天前 #Python

文章标签指南

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

什么是 webhook？

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

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

为什么使用 webhook？

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

在 django 中设置 webhook

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

第 1 步：设置 webhook url

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

在 urls.py 中：

from django.urls import pathfrom . import viewsurlpatterns = [    path("webhook/", views.payment_webhook, name="payment_webhook"),]

第 2 步：创建 webhook 视图

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

import jsonfrom django.http import jsonresponse, httpresponsebadrequestfrom django.views.decorators.csrf import csrf_exempt@csrf_exempt  # exempt this view from csrf protectiondef 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 hmacimport hashlibdef 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)

# example of celery task usagefrom .tasks import process_payment_event@csrf_exemptdef 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 可能具有挑战性，因为它们需要外部服务来触发它们。以下是一些常见的测试方法：

from django.test import TestCase, Clientimport jsonclass 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"})

结论

webhook 是创建实时事件驱动应用程序的重要组成部分，django 提供了安全有效地实现它们所需的灵活性和工具。通过遵循设计、模块化和安全性方面的最佳实践，您可以构建可扩展、可靠且有弹性的 webhook 处理。

无论是与支付处理器、社交媒体平台还是任何外部 api 集成，django 中实施良好的 webhook 系统都可以显着增强应用程序的响应能力和连接性。

文章推荐

Django 中的 Webhook：综合指南

什么是 webhook？

为什么使用 webhook？

在 django 中设置 webhook

第 1 步：设置 webhook url

第 2 步：创建 webhook 视图

第 3 步：实现辅助函数

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

webhook 的安全最佳实践

测试网络钩子

结论

Python程序示例，演示字符串插值

如何用Python编写人工神经网络算法？

如何用Python编写求解排列组合的算法？

如何用Python编写求解最小公倍数的算法？

如何使用Python实现贪心算法？