A Django command function for syncing Stripe webhook events with your local server that happened while it wasn’t running

djstripe_sync_webhook.py

import json
from traceback import format_exc

from django.conf import settings
from django.core.management.base import BaseCommand, CommandError
from djstripe import settings as djstripe_settings
from djstripe.models import Event, WebhookEventTrigger
from djstripe.signals import webhook_processing_error
# from djstripe.utils import convert_tstamp  # convert stripe timestamps to datetimes

import stripe  # this can be imported into other files!
stripe.api_key = settings.STRIPE_SECRET_KEY

# This was used in a [Django Cookiecutter project](https://cookiecutter-django.readthedocs.io/en/latest/) with Docker Compose
#
# 1. start ngrok separately and get NGROK_URL: `ngrok http localhost:3000`
# 2. run `python manage.py djstripe_sync_webhook <NGROK_URL>`
#    1. using `settings.DJSTRIPE_WEBHOOK_REF` as an identifier, delete the existing webhook if any, then create a new one
#    2. save the webhook secret to `settings.DJSTRIPE_WEBHOOK_SECRET_PATH`
#    3. query Stripe events that have happened since the last processed event and process them using djstripe
#       (NOTE: the events API only returns the last 30-days of events)
# 3. if you're running this in a container, restart django, e.g., `docker-compose -f local.yml restart django`

ENABLED_EVENTS = [
    "source.canceled",
    "source.chargeable",
    "source.failed",
    "source.mandate_notification",
    "source.refund_attributes_required",
    "source.transaction.created",
    "source.transaction.updated",
    "tax_rate.created",
    "tax_rate.updated",
    "transfer.created",
    "transfer.failed",
    "transfer.paid",
    "transfer.reversed",
    "transfer.canceled",
    "transfer.updated",
    "setup_intent.canceled",
    "setup_intent.created",
    "setup_intent.setup_failed",
    "setup_intent.succeeded",
    "product.created",
    "product.deleted",
    "product.updated",
    "plan.created",
    "plan.deleted",
    "plan.updated",
    "payment_method.attached",
    "payment_method.card_automatically_updated",
    "payment_method.detached",
    "payment_method.updated",
    "payment_intent.amount_capturable_updated",
    "payment_intent.canceled",
    "payment_intent.created",
    "payment_intent.payment_failed",
    "payment_intent.processing",
    "payment_intent.succeeded",
    "invoiceitem.created",
    "invoiceitem.deleted",
    "invoiceitem.updated",
    "invoice.created",
    "invoice.deleted",
    "invoice.finalized",
    "invoice.marked_uncollectible",
    "invoice.payment_action_required",
    "invoice.payment_failed",
    "invoice.payment_succeeded",
    "invoice.sent",
    "invoice.upcoming",
    "invoice.updated",
    "invoice.voided",
    "customer.created",
    "customer.deleted",
    "customer.updated",
    "customer.discount.created",
    "customer.discount.deleted",
    "customer.discount.updated",
    "customer.source.created",
    "customer.card.created",
    "customer.bank_account.created",
    "customer.source.deleted",
    "customer.card.deleted",
    "customer.bank_account.deleted",
    "customer.source.expiring",
    "customer.source.updated",
    "customer.card.updated",
    "customer.bank_account.updated",
    "customer.subscription.created",
    "customer.subscription.deleted",
    "customer.subscription.pending_update_applied",
    "customer.subscription.pending_update_expired",
    "customer.subscription.trial_will_end",
    "customer.subscription.updated",
    "customer.tax_id.created",
    "customer.tax_id.deleted",
    "customer.tax_id.updated",
    "coupon.created",
    "coupon.deleted",
    "coupon.updated",
    "charge.captured",
    "charge.expired",
    "charge.failed",
    "charge.pending",
    "charge.refunded",
    "charge.succeeded",
    "charge.updated",
    "charge.dispute.closed",
    "charge.dispute.created",
    "charge.dispute.funds_reinstated",
    "charge.dispute.funds_withdrawn",
    "charge.dispute.updated",
    "charge.refund.updated",
    "application_fee.created",
    "application_fee.refunded",
    "application_fee.refund.updated",
    "account.updated",
    "account.application.authorized",
    "account.application.deauthorized",
    "account.external_account.created",
    "account.external_account.deleted",
    "account.external_account.updated"
]


# # Command

class Command(BaseCommand):
    help = 'Sync events that happened while the webhook was down.'

    def add_arguments(self, parser):
        parser.add_argument('ngrok_url', type=str)
        # parser.add_argument('poll_ids', nargs='+', type=int)

    def handle(self, *args, **options):
        ngrok_url = options['ngrok_url']
        run(ngrok_url, print_fn=self.stdout.write)
        # TODO - raise CommandError when error?


# # Functions

def run(ngrok_url, print_fn=print):
    update_webhook(ngrok_url, print_fn)
    sync_events(print_fn)


# ## Get/Update webhook

def update_webhook(ngrok_url, print_fn):
    webhook_url = ngrok_url + '/connect/stripe/webhook'  # TODO(DRY) - url reverse

    webhook_ref = settings.DJSTRIPE_WEBHOOK_REF
    if not webhook_ref:
        raise CommandError(f'Must specify a DJ_STRIPE_WEBHOOK_REF in the .envs/.local/.private file')

    webhook = _get_existing_webhook(webhook_ref)

    if webhook:
        print_fn('DELETE: {webhook.id} ({webhook.url})')
        webhook.delete()

    webhook = _create_webhook(webhook_url, webhook_ref, ENABLED_EVENTS)

    _save_secret(webhook.secret)

    print_fn('\nUPDATED: Set new webhook and secret, make sure to restart docker! `docker-compose -f local.yml restart django`\n')


def _get_existing_webhook(webhook_ref):
    webhooks = stripe.WebhookEndpoint.list(limit=40)
    for webhook in webhooks.auto_paging_iter():
        if webhook.description == webhook_ref:
            return webhook


def _create_webhook(url, description, enabled_events):
    return stripe.WebhookEndpoint.create(
        url=url,
        description=description,
        enabled_events=enabled_events,
    )


def _save_secret(secret):
    with open(settings.DJSTRIPE_WEBHOOK_SECRET_PATH, 'w') as f:
        f.write(secret)


# ## Sync Events

def sync_events(print_fn):
    last_event = Event.objects.order_by('-created').first()

    kwargs = {'limit': 40}
    if last_event:
        kwargs['ending_before'] = last_event.id
    events = stripe.Event.list(**kwargs)

    for i, event in enumerate(events.auto_paging_iter()):
        # print(i, event.id, event.type)
        obj = _process_stripe_event(event)
        print_fn(f'{i+1}. Created {obj} ({event.type} - {event.created}')


def _process_stripe_event(event):
    """
    Adapted from djstripe.models.WebhookEventTrigger.from_request to insert events into our system.

    event - stripe.Event api response
    """
    body = json.dumps(event)

    obj = WebhookEventTrigger.objects.create(headers={}, body=body, remote_ip='0.0.0.0', valid=True)
    try:
        if djstripe_settings.WEBHOOK_EVENT_CALLBACK:
            # If WEBHOOK_EVENT_CALLBACK, pass it for processing
            djstripe_settings.WEBHOOK_EVENT_CALLBACK(obj)
        else:
            # Process the item (do not save it, it'll get saved below)
            obj.process(save=False)
    except Exception as e:
        max_length = WebhookEventTrigger._meta.get_field("exception").max_length
        obj.exception = str(e)[:max_length]
        obj.traceback = format_exc()

        # Send the exception as the webhook_processing_error signal
        webhook_processing_error.send(
            sender=WebhookEventTrigger,
            exception=e,
            data=getattr(e, "http_body", ""),
        )

        # re-raise the exception so Django sees it
        raise e
    finally:
        obj.save()

    return obj