August 14, 2026 · Tutorial

Add E-Signatures to Django Applications in 15 Minutes (2026)

A comprehensive, production-grade guide to integrating e-signatures in Django 5+. Learn how to build clean HTTP service objects, dispatch contracts asynchronously via Celery, and process timing-safe HMAC-SHA256 webhooks in Django REST Framework (DRF).

Michael Beckett

Founder, Signbee

TL;DR & ARCHITECTURE SUMMARY

In this tutorial, you will build an end-to-end e-signature subsystem in Django 5+. You will create a decoupled Service Layer (contracts/services/signbee.py) using httpx, dispatch document signing requests using Celery background tasks with automatic retries, store contract states and audit logs in the Django ORM, and verify incoming HMAC-SHA256 webhooks in Django REST Framework using hmac.compare_digest.

Key Stack Components & Versions (2026)

FrameworkDjango 5.0+ / 5.1

API FrameworkDjango REST Framework 3.15+

HTTP Clienthttpx 0.27+

Task QueueCelery 5.4+ (Redis/RabbitMQ)

SecurityHMAC-SHA256 Constant-Time

Database ORMPostgreSQL / Django ORM

Why Modern Django Applications Need API-First E-Signatures

Django remains the gold standard web framework for data-intensive SaaS applications, enterprise portals, HR platforms, and fintech systems. However, adding electronic signatures to Django projects has historically been a painful exercise. Legacy enterprise e-signature vendors require multi-megabyte SDKs, complex OAuth2 multi-step consent handshakes, and rigid PDF coordinate drag-and-drop builders that fail to integrate cleanly with Django's data model.

Modern web applications demand an API-first, code-driven approach:

Dynamic Markdown Document Generation: Render contract bodies directly from Django database fields and templates without needing pre-baked PDF visual templates.
Decoupled Service Layer: Encapsulate external API communications in dedicated service classes rather than cluttering views or model methods.
Non-Blocking Background Tasks: Execute outbound HTTP calls inside Celery workers, keeping WSGI/ASGI web processes responsive and resilient.
Cryptographic Webhook Handlers: Securely process contract state transitions (sent, viewed, signed, declined) using timing-safe HMAC verification and atomic database transactions.

If you are building with FastAPI or Ruby on Rails instead, check out our companion guides on FastAPI E-Signature Integration, Python E-Signature API Integration (Requests & Flask), and Ruby on Rails E-Signature API Guide.

Architecture: Django Service Layer & Celery Workflow

Before writing code, let's examine the system architecture and lifecycle of an e-signature transaction in a production Django environment:

End-to-End Django E-Signature Flow

1.User / App creates Contract model in Django (status: DRAFT).
2.Django view triggers transaction.on_commit(lambda: send_contract_task.delay(contract.id)).
3.Celery worker picks up task, calls SignbeeService.create_signature_request() via httpx.
4.Signbee returns document_id and signing_url. Celery task updates Contract to PENDING.
5.Recipient receives email or opens embedded signing iframe and executes the signature.
6.Signbee issues an HTTP POST webhook with X-Signbee-Signature header.
7.DRF Webhook view verifies HMAC-SHA256 signature using hmac.compare_digest and atomically updates model to SIGNED.

Step 1: Installing Dependencies and Configuring Django Settings

Start by installing the necessary Python packages in your Django virtual environment:

Terminal — install dependenciesPython / Django 5+

pip install django djangorestframework httpx celery redis python-dotenv

Next, add your Signbee API credentials to your .env file:

.envPython / Django 5+

SIGNBEE_API_KEY=sb_live_a8f93b2c1d0e4f5a6b7c8d9e0f1a2b3c
SIGNBEE_WEBHOOK_SECRET=whsec_99a88b77c66d55e44f33a22b11c00d9e
SIGNBEE_BASE_URL=https://api.signb.ee/v1
CELERY_BROKER_URL=redis://localhost:6379/0

In your Django settings.py, register rest_framework and your contracts app, and expose the environment variables:

core/settings.pyPython / Django 5+

import os
from pathlib import Path
from dotenv import load_dotenv

load_dotenv()

BASE_DIR = Path(__file__).resolve().parent.parent

INSTALLED_APPS = [
    'django.contrib.admin',
    'django.contrib.auth',
    'django.contrib.contenttypes',
    'django.contrib.sessions',
    'django.contrib.messages',
    'django.contrib.staticfiles',
    
    # Third-party apps
    'rest_framework',
    
    # Local apps
    'contracts.apps.ContractsConfig',
]

# Signbee E-Signature API Configuration
SIGNBEE_API_KEY = os.getenv("SIGNBEE_API_KEY")
SIGNBEE_WEBHOOK_SECRET = os.getenv("SIGNBEE_WEBHOOK_SECRET")
SIGNBEE_BASE_URL = os.getenv("SIGNBEE_BASE_URL", "https://api.signb.ee/v1")
SIGNBEE_TIMEOUT_SECONDS = float(os.getenv("SIGNBEE_TIMEOUT_SECONDS", "15.0"))

# Celery Configuration
CELERY_BROKER_URL = os.getenv("CELERY_BROKER_URL", "redis://localhost:6379/0")
CELERY_RESULT_BACKEND = os.getenv("CELERY_RESULT_BACKEND", "redis://localhost:6379/0")
CELERY_ACCEPT_CONTENT = ["json"]
CELERY_TASK_SERIALIZER = "json"
CELERY_RESULT_SERIALIZER = "json"
CELERY_TIMEZONE = "UTC"

Configure your Celery application in core/celery.py:

core/celery.pyPython / Django 5+

import os
from celery import Celery

os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'core.settings')

app = Celery('core')
app.config_from_object('django.conf:settings', namespace='CELERY')
app.autodiscover_tasks()

And ensure Celery is loaded in core/__init__.py:

core/__init__.pyPython / Django 5+

from .celery import app as celery_app

__all__ = ('celery_app',)

Step 2: Defining the Django ORM Data Model

A robust e-signature integration requires tracking contract lifecycle states, signer records, external API identifiers, and an immutable audit log. Let's create contracts/models.py:

contracts/models.pyPython / Django 5+

import uuid
from django.db import models
from django.utils import timezone


class ContractStatus(models.TextChoices):
    DRAFT = "DRAFT", "Draft"
    QUEUED = "QUEUED", "Queued for Dispatch"
    PENDING = "PENDING", "Pending Signature"
    VIEWED = "VIEWED", "Viewed by Recipient"
    SIGNED = "SIGNED", "Signed & Completed"
    DECLINED = "DECLINED", "Declined"
    EXPIRED = "EXPIRED", "Expired"
    ERROR = "ERROR", "Dispatch Error"


class Contract(models.Model):
    id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False)
    title = models.CharField(max_length=255, help_text="Human-readable agreement title")
    markdown_content = models.TextField(help_text="Dynamic Markdown content of the contract")
    
    # Recipient details
    recipient_name = models.CharField(max_length=255)
    recipient_email = models.EmailField()
    
    # Signbee API references
    signbee_document_id = models.CharField(
        max_length=128, 
        blank=True, 
        null=True, 
        unique=True, 
        db_index=True,
        help_text="Unique document ID returned by Signbee API"
    )
    signing_url = models.URLField(
        max_length=1024, 
        blank=True, 
        null=True, 
        help_text="Hosted or embedded signing URL"
    )
    signed_pdf_url = models.URLField(
        max_length=1024, 
        blank=True, 
        null=True, 
        help_text="Permanent storage URL of the completed signed PDF"
    )
    audit_certificate_url = models.URLField(
        max_length=1024, 
        blank=True, 
        null=True, 
        help_text="Cryptographic audit trail certificate URL"
    )
    
    # Status and timestamps
    status = models.CharField(
        max_length=32, 
        choices=ContractStatus.choices, 
        default=ContractStatus.DRAFT,
        db_index=True
    )
    error_message = models.TextField(blank=True, null=True)
    created_at = models.DateTimeField(auto_now_add=True)
    updated_at = models.DateTimeField(auto_now=True)
    sent_at = models.DateTimeField(blank=True, null=True)
    signed_at = models.DateTimeField(blank=True, null=True)

    class Meta:
        ordering = ["-created_at"]
        verbose_name = "Contract"
        verbose_name_plural = "Contracts"

    def __str__(self):
        return f"{self.title} ({self.recipient_name}) - [{self.status}]"


class ContractAuditLog(models.Model):
    id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False)
    contract = models.ForeignKey(
        Contract, 
        on_delete=models.CASCADE, 
        related_name="audit_logs"
    )
    event_type = models.CharField(max_length=64, db_index=True)
    event_payload = models.JSONField(help_text="Raw JSON payload received from webhook")
    ip_address = models.GenericIPAddressField(blank=True, null=True)
    user_agent = models.TextField(blank=True, null=True)
    created_at = models.DateTimeField(default=timezone.now)

    class Meta:
        ordering = ["-created_at"]
        verbose_name = "Contract Audit Log"
        verbose_name_plural = "Contract Audit Logs"

    def __str__(self):
        return f"{self.contract.title} - {self.event_type} at {self.created_at}"

Create and execute the database migrations:

Terminal — run migrationsPython / Django 5+

python manage.py makemigrations contracts
python manage.py migrate

Step 3: Building the Signbee Service Layer

Rather than calling external HTTP endpoints directly within views or tasks, we implement a dedicated service class in contracts/services/signbee.py using httpx. This encapsulates authentication, timeout configuration, error handling, and payload construction.

contracts/services/signbee.pyPython / Django 5+

from typing import Dict, Any, Optional
import httpx
import logging
from django.conf import settings

logger = logging.getLogger(__name__)


class SignbeeAPIError(Exception):
    """Custom exception raised when Signbee API returns an error response."""
    def __init__(self, message: str, status_code: Optional[int] = None, response_body: Optional[str] = None):
        super().__init__(message)
        self.status_code = status_code
        self.response_body = response_body


class SignbeeService:
    """
    Encapsulates all outbound REST API communication with Signbee.
    Uses httpx for modern HTTP/2 support, custom timeouts, and structured error handling.
    """

    def __init__(
        self, 
        api_key: Optional[str] = None, 
        base_url: Optional[str] = None,
        timeout: Optional[float] = None
    ):
        self.api_key = api_key or settings.SIGNBEE_API_KEY
        self.base_url = (base_url or settings.SIGNBEE_BASE_URL).rstrip("/")
        self.timeout = timeout or settings.SIGNBEE_TIMEOUT_SECONDS
        
        if not self.api_key:
            raise ValueError("Signbee API key is not configured in Django settings.")

    @property
    def _headers(self) -> Dict[str, str]:
        return {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json",
            "Accept": "application/json",
            "User-Agent": "Signbee-Django-Integration/2026.1",
        }

    def create_signature_request(
        self,
        title: str,
        markdown_content: str,
        recipient_name: str,
        recipient_email: str,
        external_id: Optional[str] = None,
        metadata: Optional[Dict[str, Any]] = None,
    ) -> Dict[str, Any]:
        """
        Creates and dispatches a document signing request via Signbee API.
        
        Returns:
            Dict containing 'document_id', 'signing_url', 'status', etc.
        """
        url = f"{self.base_url}/documents"
        payload = {
            "title": title,
            "content": markdown_content,
            "recipients": [
                {
                    "name": recipient_name,
                    "email": recipient_email,
                    "role": "signer",
                }
            ],
            "external_id": external_id,
            "metadata": metadata or {},
        }

        try:
            with httpx.Client(timeout=self.timeout) as client:
                response = client.post(url, json=payload, headers=self._headers)
                
                if response.is_error:
                    logger.error(
                        f"Signbee API Error ({response.status_code}): {response.text}"
                    )
                    raise SignbeeAPIError(
                        message=f"Signbee API request failed: {response.status_code}",
                        status_code=response.status_code,
                        response_body=response.text,
                    )
                
                return response.json()

        except httpx.RequestError as exc:
            logger.error(f"Network error connecting to Signbee API: {exc}")
            raise SignbeeAPIError(f"HTTP network error: {str(exc)}") from exc

    def get_document_status(self, document_id: str) -> Dict[str, Any]:
        """Fetches the current real-time status and metadata of a document."""
        url = f"{self.base_url}/documents/{document_id}"
        try:
            with httpx.Client(timeout=self.timeout) as client:
                response = client.get(url, headers=self._headers)
                response.raise_for_status()
                return response.json()
        except httpx.HTTPError as exc:
            raise SignbeeAPIError(f"Failed to fetch document {document_id}: {str(exc)}") from exc

    def download_signed_pdf(self, document_id: str) -> bytes:
        """Downloads the completed, cryptographically signed PDF binary."""
        url = f"{self.base_url}/documents/{document_id}/download"
        try:
            with httpx.Client(timeout=self.timeout) as client:
                response = client.get(url, headers=self._headers)
                response.raise_for_status()
                return response.content
        except httpx.HTTPError as exc:
            raise SignbeeAPIError(f"Failed to download PDF for {document_id}: {str(exc)}") from exc

Step 4: Asynchronous Celery Dispatch Task with Retries

Network calls should never block Django web request threads. We create a Celery shared task in contracts/tasks.py that handles exponential backoff, transaction locks with select_for_update(), and status tracking:

contracts/tasks.pyPython / Django 5+

import logging
from celery import shared_task
from django.db import transaction
from django.utils import timezone
from .models import Contract, ContractStatus
from .services.signbee import SignbeeService, SignbeeAPIError

logger = logging.getLogger(__name__)


@shared_task(
    bind=True,
    max_retries=4,
    default_retry_delay=60,
    autoretry_for=(SignbeeAPIError,),
    retry_backoff=True,
    retry_backoff_max=600,
    retry_jitter=True,
)
def send_contract_task(self, contract_id: str) -> str:
    """
    Asynchronously dispatches a Contract to Signbee API.
    Retries automatically with exponential backoff if transient errors occur.
    """
    logger.info(f"Starting e-signature dispatch for Contract {contract_id} (Attempt {self.request.retries + 1})")

    # Fetch contract with atomic row locking
    with transaction.atomic():
        try:
            contract = Contract.objects.select_for_update().get(id=contract_id)
        except Contract.DoesNotExist:
            logger.error(f"Contract {contract_id} not found. Aborting task.")
            return "Contract not found"

        if contract.status in [ContractStatus.PENDING, ContractStatus.SIGNED]:
            logger.warning(f"Contract {contract_id} is already {contract.status}. Skipping dispatch.")
            return f"Already {contract.status}"

        contract.status = ContractStatus.QUEUED
        contract.save(update_fields=["status", "updated_at"])

    # Call Signbee Service outside DB lock to prevent holding open database connections
    service = SignbeeService()
    try:
        response_data = service.create_signature_request(
            title=contract.title,
            markdown_content=contract.markdown_content,
            recipient_name=contract.recipient_name,
            recipient_email=contract.recipient_email,
            external_id=str(contract.id),
            metadata={"django_contract_id": str(contract.id)},
        )

        document_id = response_data.get("id") or response_data.get("document_id")
        signing_url = response_data.get("signing_url")

        with transaction.atomic():
            contract.refresh_from_db()
            contract.signbee_document_id = document_id
            contract.signing_url = signing_url
            contract.status = ContractStatus.PENDING
            contract.sent_at = timezone.now()
            contract.error_message = None
            contract.save(update_fields=[
                "signbee_document_id", 
                "signing_url", 
                "status", 
                "sent_at", 
                "error_message", 
                "updated_at"
            ])

        logger.info(f"Successfully dispatched Contract {contract_id} (Signbee Doc ID: {document_id})")
        return f"Dispatched: {document_id}"

    except SignbeeAPIError as exc:
        logger.error(f"Signbee API failure on Contract {contract_id}: {exc}")
        with transaction.atomic():
            contract.refresh_from_db()
            contract.status = ContractStatus.ERROR
            contract.error_message = str(exc)
            contract.save(update_fields=["status", "error_message", "updated_at"])
        
        # Trigger Celery retry with backoff
        raise self.retry(exc=exc)

Best Practice: Triggering Celery in Views with transaction.on_commit

Always enqueue Celery tasks using transaction.on_commit(). This ensures that the database transaction saving your Contract record is fully committed to PostgreSQL before the Celery worker attempts to read it:

contracts/views.py (Contract Creation Example)Python / Django 5+

from django.db import transaction
from django.shortcuts import redirect
from .models import Contract
from .tasks import send_contract_task

def dispatch_new_contract_view(request):
    with transaction.atomic():
        contract = Contract.objects.create(
            title="Non-Disclosure Agreement (2026)",
            recipient_name="Alice Smith",
            recipient_email="alice@example.com",
            markdown_content="# NDA\n\nThis Non-Disclosure Agreement...",
        )
        # Guarantees task only executes AFTER the DB transaction commits
        transaction.on_commit(lambda: send_contract_task.delay(str(contract.id)))

    return redirect("contract-detail", pk=contract.id)

Step 5: Django REST Framework Webhook View with Constant-Time HMAC Verification

When recipients open, sign, or decline a document, Signbee sends an HTTP POST notification to your registered webhook URL. To prevent spoofing and tampering, each payload includes the X-Signbee-Signature header containing an HMAC-SHA256 digest computed with your webhook secret.

In Python and Django, you must:

Verify against the raw, unparsed request bytes (request.body).
Use hmac.compare_digest for constant-time comparison to prevent timing attacks.
Maintain idempotency so duplicate webhook deliveries do not corrupt state.

contracts/views.pyPython / Django 5+

import hmac
import hashlib
import json
import logging
from django.conf import settings
from django.db import transaction
from django.utils import timezone
from rest_framework.views import APIView
from rest_framework.response import Response
from rest_framework import status
from .models import Contract, ContractStatus, ContractAuditLog

logger = logging.getLogger(__name__)


class SignbeeWebhookView(APIView):
    """
    Receives and processes incoming Signbee webhook event notifications.
    Validates HMAC-SHA256 signatures in constant time.
    """
    authentication_classes = []  # Public endpoint verified via HMAC
    permission_classes = []

    def post(self, request, *args, **kwargs):
        # 1. Retrieve the incoming signature header
        signature_header = request.headers.get("X-Signbee-Signature") or request.headers.get("x-signbee-signature")
        if not signature_header:
            logger.warning("Webhook request missing X-Signbee-Signature header")
            return Response({"error": "Missing signature header"}, status=status.HTTP_401_UNAUTHORIZED)

        # 2. Extract raw request bytes before any JSON deserialization
        raw_body = request.body
        secret = settings.SIGNBEE_WEBHOOK_SECRET
        if not secret:
            logger.error("SIGNBEE_WEBHOOK_SECRET is not configured in Django settings.")
            return Response({"error": "Server misconfiguration"}, status=status.HTTP_500_INTERNAL_SERVER_ERROR)

        # 3. Compute expected HMAC-SHA256 digest
        computed_hmac = hmac.new(
            key=secret.encode("utf-8"),
            msg=raw_body,
            digestmod=hashlib.sha256
        ).hexdigest()

        # Handle 'sha256=' prefix if present in the header
        provided_signature = signature_header.replace("sha256=", "").strip()

        # 4. Constant-time comparison to eliminate timing attacks
        if not hmac.compare_digest(computed_hmac, provided_signature):
            logger.warning("HMAC signature verification failed for incoming webhook.")
            return Response({"error": "Invalid signature"}, status=status.HTTP_401_UNAUTHORIZED)

        # 5. Parse JSON payload
        try:
            payload = json.loads(raw_body.decode("utf-8"))
        except (ValueError, UnicodeDecodeError) as exc:
            logger.error(f"Malformed JSON in webhook body: {exc}")
            return Response({"error": "Invalid JSON"}, status=status.HTTP_400_BAD_REQUEST)

        event_type = payload.get("event") or payload.get("type")
        document_data = payload.get("data", {})
        document_id = document_data.get("id") or document_data.get("document_id")

        if not document_id:
            logger.warning("Webhook payload missing document ID")
            return Response({"error": "Missing document ID in payload"}, status=status.HTTP_400_BAD_REQUEST)

        logger.info(f"Processing verified Signbee webhook '{event_type}' for document '{document_id}'")

        # 6. Apply state transitions atomically
        with transaction.atomic():
            try:
                contract = Contract.objects.select_for_update().get(signbee_document_id=document_id)
            except Contract.DoesNotExist:
                # Also check if external_id was passed
                external_id = document_data.get("external_id")
                try:
                    contract = Contract.objects.select_for_update().get(id=external_id)
                except (Contract.DoesNotExist, ValueError):
                    logger.warning(f"No contract matching Signbee Doc ID '{document_id}' or external_id '{external_id}'")
                    # Return 200 to acknowledge webhook even if contract not in local DB
                    return Response({"status": "ignored_unknown_contract"}, status=status.HTTP_200_OK)

            # Record immutable audit log
            ContractAuditLog.objects.create(
                contract=contract,
                event_type=event_type,
                event_payload=payload,
                ip_address=request.META.get("REMOTE_ADDR"),
                user_agent=request.META.get("HTTP_USER_AGENT", "")[:500],
            )

            # Handle event transitions
            if event_type == "contract.viewed" or event_type == "document.viewed":
                if contract.status == ContractStatus.PENDING:
                    contract.status = ContractStatus.VIEWED
                    contract.save(update_fields=["status", "updated_at"])

            elif event_type == "contract.signed" or event_type == "document.signed":
                contract.status = ContractStatus.SIGNED
                contract.signed_at = timezone.now()
                contract.signed_pdf_url = document_data.get("download_url") or document_data.get("signed_pdf_url")
                contract.audit_certificate_url = document_data.get("audit_certificate_url")
                contract.save(update_fields=[
                    "status", 
                    "signed_at", 
                    "signed_pdf_url", 
                    "audit_certificate_url", 
                    "updated_at"
                ])
                logger.info(f"Contract {contract.id} marked as SIGNED.")

            elif event_type == "contract.declined" or event_type == "document.declined":
                contract.status = ContractStatus.DECLINED
                contract.error_message = document_data.get("decline_reason", "Declined by recipient")
                contract.save(update_fields=["status", "error_message", "updated_at"])

            elif event_type == "contract.expired" or event_type == "document.expired":
                contract.status = ContractStatus.EXPIRED
                contract.save(update_fields=["status", "updated_at"])

        return Response({"status": "success", "event": event_type}, status=status.HTTP_200_OK)

Step 6: URL Routing & Django Admin Interface

Wire up the webhook view in contracts/urls.py and include it in your project's main URL configuration:

contracts/urls.pyPython / Django 5+

from django.urls import path
from .views import SignbeeWebhookView

app_name = "contracts"

urlpatterns = [
    path("webhooks/signbee/", SignbeeWebhookView.as_view(), name="signbee-webhook"),
]

Next, configure contracts/admin.py to inspect contracts, statuses, signing links, and audit logs directly from the Django Admin:

contracts/admin.pyPython / Django 5+

from django.contrib import admin
from django.utils.html import format_html
from .models import Contract, ContractAuditLog


class ContractAuditLogInLine(admin.TabularInline):
    model = ContractAuditLog
    extra = 0
    readonly_fields = ("event_type", "ip_address", "created_at", "event_payload")
    can_delete = False


@admin.register(Contract)
class ContractAdmin(admin.ModelAdmin):
    list_display = (
        "title", 
        "recipient_name", 
        "recipient_email", 
        "status_badge", 
        "sent_at", 
        "signed_at", 
        "view_signing_link"
    )
    list_filter = ("status", "created_at", "signed_at")
    search_fields = ("title", "recipient_name", "recipient_email", "signbee_document_id")
    readonly_fields = (
        "id", 
        "signbee_document_id", 
        "signing_url", 
        "signed_pdf_url", 
        "audit_certificate_url", 
        "created_at", 
        "updated_at"
    )
    inlines = [ContractAuditLogInLine]

    def status_badge(self, obj):
        colors = {
            "DRAFT": "#6b7280",
            "QUEUED": "#f59e0b",
            "PENDING": "#3b82f6",
            "VIEWED": "#8b5cf6",
            "SIGNED": "#10b981",
            "DECLINED": "#ef4444",
            "ERROR": "#dc2626",
        }
        color = colors.get(obj.status, "#6b7280")
        return format_html(
            '<span style="background-color: {}; color: white; padding: 2px 8px; border-radius: 4px; font-weight: bold; font-size: 11px;">{}</span>',
            color,
            obj.get_status_display(),
        )
    status_badge.short_description = "Status"

    def view_signing_link(self, obj):
        if obj.signing_url:
            return format_html('<a href="{}" target="_blank" rel="noopener">Sign Link ↗</a>', obj.signing_url)
        return "-"
    view_signing_link.short_description = "Signing Link"

Step 7: Automated Unit & Integration Testing

Writing rigorous automated tests guarantees that your HMAC verification and Celery dispatch routines remain rock-solid across updates. Here is a complete Django test suite in contracts/tests/test_webhooks.py:

contracts/tests/test_webhooks.pyPython / Django 5+

import hmac
import hashlib
import json
from django.test import TestCase, override_settings
from django.urls import reverse
from rest_framework import status
from contracts.models import Contract, ContractStatus, ContractAuditLog


@override_settings(SIGNBEE_WEBHOOK_SECRET="test_webhook_secret_key_12345")
class SignbeeWebhookTests(TestCase):
    def setUp(self):
        self.webhook_url = reverse("contracts:signbee-webhook")
        self.secret = "test_webhook_secret_key_12345"
        self.contract = Contract.objects.create(
            title="Master Services Agreement",
            recipient_name="John Doe",
            recipient_email="john@example.com",
            markdown_content="# MSA Agreement",
            signbee_document_id="doc_test_abc123",
            status=ContractStatus.PENDING,
        )

    def _generate_signature(self, payload_bytes: bytes) -> str:
        return hmac.new(
            key=self.secret.encode("utf-8"),
            msg=payload_bytes,
            digestmod=hashlib.sha256
        ).hexdigest()

    def test_valid_signature_updates_contract_to_signed(self):
        payload = {
            "event": "contract.signed",
            "data": {
                "id": "doc_test_abc123",
                "signed_pdf_url": "https://storage.signb.ee/signed/doc_test_abc123.pdf",
                "audit_certificate_url": "https://storage.signb.ee/certs/doc_test_abc123.pdf",
            }
        }
        body_bytes = json.dumps(payload).encode("utf-8")
        signature = self._generate_signature(body_bytes)

        response = self.client.post(
            self.webhook_url,
            data=body_bytes,
            content_type="application/json",
            HTTP_X_SIGNBEE_SIGNATURE=signature,
        )

        self.assertEqual(response.status_code, status.HTTP_200_OK)
        self.contract.refresh_from_db()
        self.assertEqual(self.contract.status, ContractStatus.SIGNED)
        self.assertIsNotNone(self.contract.signed_at)
        self.assertEqual(self.contract.signed_pdf_url, "https://storage.signb.ee/signed/doc_test_abc123.pdf")
        
        # Verify audit log was recorded
        self.assertEqual(ContractAuditLog.objects.filter(contract=self.contract).count(), 1)

    def test_invalid_signature_rejected_with_401(self):
        payload = {"event": "contract.signed", "data": {"id": "doc_test_abc123"}}
        body_bytes = json.dumps(payload).encode("utf-8")

        response = self.client.post(
            self.webhook_url,
            data=body_bytes,
            content_type="application/json",
            HTTP_X_SIGNBEE_SIGNATURE="tampered_invalid_signature_hash",
        )

        self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED)
        self.contract.refresh_from_db()
        self.assertEqual(self.contract.status, ContractStatus.PENDING)

Run your test suite with manage.py or pytest:

Terminal — execute testsPython / Django 5+

python manage.py test contracts

Production Deployment & Reliability Checklist

When taking your Django e-signature integration to production, verify the following operational configurations:

Celery Worker Concurrency & Dead Letter Queues

Allocate dedicated Celery queues for external I/O tasks (e.g. celery -A core worker -Q esignatures,default -c 4). Ensure failed retries route to a Dead Letter Queue for monitoring and alerting.

Database Connection Pooling & Transactions

Use CONN_MAX_AGE or PgBouncer in transaction pooling mode. Always place outbound network requests outside of atomic database transaction blocks to prevent holding open database connections during network latency spikes.

Archiving Signed PDFs to S3 / Cloud Storage

Upon receiving the contract.signed webhook, trigger a secondary Celery task that calls SignbeeService.download_signed_pdf() and saves the immutable binary to AWS S3, Google Cloud Storage, or Azure Blob Storage via django-storages.

Frequently Asked Questions

Why should Django applications dispatch e-signature requests via Celery background tasks instead of views?

Executing outbound HTTP requests synchronously inside standard Django view functions or Gunicorn/uWSGI worker processes creates severe performance and reliability bottlenecks. Generating document content, contacting external e-signature REST endpoints, and awaiting cryptographic token generation can take anywhere from 300ms to several seconds depending on network conditions. In synchronous WSGI architectures, blocking an HTTP worker thread quickly exhausts the worker pool during traffic spikes, causing request queuing, timeout errors, and sluggish response times for all site visitors. By delegating contract dispatching to asynchronous Celery background tasks backed by Redis or RabbitMQ, Django views return immediate HTTP responses (such as HTTP 202 Accepted) to users. Furthermore, Celery provides native retry mechanisms with exponential backoff, automatic jitter, failure logging, and transaction isolation (transaction.on_commit), ensuring that temporary third-party network blips do not result in failed or abandoned customer transactions.

How does hmac.compare_digest protect Django webhook endpoints against timing attacks?

Standard Python string equality operators (==) evaluate strings character by character and terminate immediately upon encountering the first mismatching character. When validating cryptographic signatures like HMAC-SHA256, this short-circuiting behavior introduces measurable nanosecond latency differences based on how many leading characters of a forged signature match the legitimate digest. Malicious attackers can exploit these minuscule timing discrepancies through high-frequency statistical analysis to deduce the authentic signature byte-by-byte without knowing the shared webhook secret. Python's built-in hmac.compare_digest function eliminates this vulnerability by executing a constant-time comparison algorithm that always inspects every single byte of both strings before returning a boolean result. Using hmac.compare_digest on the unparsed raw request.body bytes ensures your Django REST Framework webhook listener remains completely immune to timing-attack vectors while verifying payload authenticity and integrity.

How should Django manage race conditions between Celery tasks and immediate incoming webhooks?

When a user signs an e-signature document immediately upon generation (or in embedded signing workflows), the Signbee webhook callback notification might hit your Django application before the initial Celery dispatch task has finished committing the new document ID to PostgreSQL. To prevent race conditions, stale overwrites, or missing foreign key errors, Django architectures should apply three best practices: First, dispatch Celery tasks strictly inside transaction.on_commit() hooks so background workers never execute before database records are fully committed. Second, utilize Django ORM's select_for_update() within an atomic transaction (transaction.atomic()) inside webhook handlers to acquire row-level locks on the contract record before applying status transitions. Third, design webhook handlers to be strictly idempotent, logging incoming events to an immutable ContractAuditLog table and ignoring redundant state transitions if the contract has already reached a terminal state such as SIGNED or VOIDED.

Ready to integrate fast, developer-friendly e-signatures into your Django project? Get started with 5 free documents/month.

Explore API Documentation FastAPI Guide →

Last updated: August 14, 2026 · Michael Beckett is the founder of Signbee and B2bee Ltd.