Django 中的 Webhook：綜合指南-Python教學-PHP中文網

Webhooks in Django: A Comprehensive Guide

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 步：實現輔助函數

為了保持視圖的簡潔和模組化，最好建立單獨的函數來處理每個特定的事件類型。

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)

登入後複製