Skip to content

Instantly share code, notes, and snippets.

Embed
What would you like to do?
A Django command function for syncing Stripe webhook events with your local server that happened while it wasn’t running
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
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment