Skip to content

DocuSign Webhook

Route & Auth

Property Value
Method POST
Path /webhooks/docusign
Auth type sha256_base64
Header x-docusign-signature-1
Secret DOCUSIGN_WEBHOOK_SECRET_NAME
Service account docusign-webhook@alan-eu-tools.iam.gserviceaccount.com
Response 200 OK

Payload

Field Type Description
event string Envelope event type
data.envelopeId string Envelope identifier
data.envelopeSummary dict Full envelope details

Recognized events

Event Description
envelope-sent Envelope sent to recipients
envelope-delivered Envelope delivered to recipients
envelope-completed All recipients have signed
envelope-voided Envelope voided by sender
envelope-declined Recipient declined to sign

Behavior

  1. Validates the event is one of the recognized types
  2. Enqueues process_envelope_updated background job
  3. The job finds the DocusignEnvelope by envelopeId
  4. Updates the appropriate timestamp field (sent_at, delivered_at, signed_at, voided_at, declined_at)
  5. On envelope-completed: downloads the signed documents archive from DocuSign and uploads to Google Drive
flowchart TD
    A[POST /webhooks/docusign] --> B{Known event?}
    B -->|No| C[Ignore]
    B -->|Yes| D[Enqueue process_envelope_updated]
    D --> E[Find DocusignEnvelope by ID]
    E --> F[Update timestamp field]
    F --> G{envelope-completed?}
    G -->|No| H[Done]
    G -->|Yes| I[Download signed docs from DocuSign]
    I --> J[Upload to Google Drive]
    J --> H
Hold "Alt" / "Option" to enable pan & zoom

Integrations

  • DocuSign API: document download
  • Google Drive API: signed document archival

Code reference

apps.eu_tools.webhooks.docusign.DocusignWebhook

Bases: MethodView

post

post(payload)
Source code in apps/eu_tools/webhooks/docusign.py
@docusign_blueprint.arguments(DocusignWebhookSchema)
@docusign_blueprint.response(status_code=HTTPStatus.OK)
def post(self, payload):  # type: ignore[no-untyped-def]
    if payload["event"] in (
        "envelope-sent",
        "envelope-delivered",
        "envelope-completed",
        "envelope-voided",
        "envelope-declined",
    ):
        current_rq.get_queue(DEFAULT_PRIORITY).enqueue(
            process_envelope_updated, payload["event"], payload["data"]
        )

    return

apps.eu_tools.webhooks.docusign.DocusignWebhookSchema

Bases: Schema

Meta

unknown class-attribute instance-attribute

unknown = EXCLUDE

data class-attribute instance-attribute

data = Dict(required=True)

event class-attribute instance-attribute

event = String(required=True)

apps.eu_tools.webhooks.docusign.docusign_blueprint module-attribute

docusign_blueprint = CustomBlueprint(
    "docusign",
    "docusign",
    url_prefix="/webhooks/docusign",
    auth_context_providers=[
        WebhookAuthContextProvider(
            auth_type=sha256_base64,
            header_name="x-docusign-signature-1",
            secret_name_config_key="DOCUSIGN_WEBHOOK_SECRET_NAME",
            auth_principal_type=ServiceAccount,
            auth_principal_email="docusign-webhook@alan-eu-tools.iam.gserviceaccount.com",
        )
    ],
)

apps.eu_tools.webhooks.docusign.process_envelope_updated

process_envelope_updated(event, envelope_summary)
Source code in apps/eu_tools/webhooks/docusign.py
@enqueueable
def process_envelope_updated(event: str, envelope_summary: dict) -> None:  # type: ignore[type-arg]
    docusign_envelope = (
        current_session.execute(
            select(DocusignEnvelope).filter(
                DocusignEnvelope.docusign_envelope_id == envelope_summary["envelopeId"]
            )
        )
        .scalars()
        .unique()
        .one_or_none()
    )
    if docusign_envelope is None:
        current_logger.info(
            "Could not find DocusignEnvelope",
            docusign_envelope_id=envelope_summary["envelopeId"],
        )
        return

    if (
        docusign_envelope.signed_at is not None
        or docusign_envelope.voided_at is not None
        or docusign_envelope.declined_at is not None
    ):
        current_logger.info(
            "DocusignEnvelope is already signed or voided or declined",
            docusign_envelope_id=envelope_summary["envelopeId"],
        )
        return

    if event == "envelope-sent":
        envelope_sent_at = datetime.fromisoformat(
            envelope_summary["envelopeSummary"]["sentDateTime"]
        )
        docusign_envelope.sent_at = envelope_sent_at
    elif event == "envelope-delivered":
        envelope_delivered_at = datetime.fromisoformat(
            envelope_summary["envelopeSummary"]["deliveredDateTime"]
        )
        docusign_envelope.delivered_at = envelope_delivered_at
    elif event == "envelope-completed":
        envelope_completed_at = datetime.fromisoformat(
            envelope_summary["envelopeSummary"]["completedDateTime"]
        )
        docusign_envelope.signed_at = envelope_completed_at
    elif event == "envelope-voided":
        envelope_voided_at = datetime.fromisoformat(
            envelope_summary["envelopeSummary"]["voidedDateTime"]
        )
        docusign_envelope.voided_at = envelope_voided_at
    elif event == "envelope-declined":
        envelope_declined_at = datetime.fromisoformat(
            envelope_summary["envelopeSummary"]["declinedDateTime"]
        )
        docusign_envelope.declined_at = envelope_declined_at

    current_session.commit()

    if event == "envelope-completed":
        alaner_profile = (
            current_session.execute(
                select(AlanerProfile)
                .join(AlanerProfile.signed_documents)
                .join(SignedDocument.docusign_envelope)
                .filter(DocusignEnvelope.id == docusign_envelope.id)
            )
            .scalars()
            .unique()
            .one_or_none()
        )
        if alaner_profile is None:
            current_logger.error(
                "Could not find AlanerProfile linked to DocusignEnvelope",
                docusign_envelope_id=docusign_envelope.id,
            )
            return
        archive_data = DocusignClient().get_envelope_documents_archive(
            envelope_summary["envelopeId"]
        )
        drive_service = get_drive_service()
        with BytesIO(archive_data) as archive_file:
            create_file(
                drive_service,
                mandatory(alaner_profile.restricted_drive_directory_id),
                f"{envelope_summary['envelopeSummary']['emailSubject']}.zip",
                archive_file,
                "application/zip",
                ignore_default_visibility=False,
            )

    current_logger.info(
        f"Processed completed envelope {envelope_summary['envelopeId']} - {event}"
    )

OpenAPI

DocuSign webhook on ReDoc ⧉