-
Notifications
You must be signed in to change notification settings - Fork 5
/
webhooks.py
84 lines (66 loc) · 2.49 KB
/
webhooks.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
import os
import time
import stripe
from django.core.mail import send_mail
from django.shortcuts import get_object_or_404
from .models import Event, Customer, Subscription
from .events import get_customer_from_event
stripe.api_key = os.environ.get("STRIPE_SECRET")
def customer_created(event):
# print("processing customer.created event...")
# customer = Customer.objects.get(stripe_id=event.data["object"]["id"])
# send_mail(
# "Payment method updated",
# "Your payment method was set successfully! Please write back if you have any questions.",
# "notifications@compute.studio",
# [customer.user.email],
# fail_silently=False,
# )
pass
def customer_subscription_deleted(event):
print("processing customer.subscription.deleted event...")
sub = get_object_or_404(Subscription, stripe_id=event.data.object.id)
customer = sub.customer
sub.delete()
print(
"successfully deleted subscription",
customer.stripe_id,
customer.user.email,
event.data.object.id,
)
# send_mail(
# "Your C/S subscription has been cancelled",
# (
# "We are sorry to see you go. If you have a moment, please let us know why "
# "you have cancelled your subscription and what we can do to win you back "
# "in the future.\n\nBest,\nThe C/S Team"
# ),
# "admin@compute.studio",
# [customer.user.email],
# fail_silently=False,
# )
def customer_subscription_updated(event: stripe.Event):
print("processing customer.subscription.updated event...")
stripe_sub: stripe.Subscription = event.data.object
sub: Subscription = get_object_or_404(Subscription, stripe_id=stripe_sub.id)
sub.update_from_stripe_obj(stripe_sub)
webhook_map = {
"customer.created": customer_created,
"customer.subscription.deleted": customer_subscription_deleted,
"customer.subscription.updated": customer_subscription_updated,
}
def process_event(stripe_event):
start = time.time()
print("got event: ")
print(stripe_event)
customer = get_customer_from_event(stripe_event)
event = Event.construct(stripe_event, customer=customer)
func = webhook_map.get(event.kind, None)
if func is not None:
try:
func(event)
except Exception as e:
import traceback
traceback.print_exc()
total_time = time.time() - start
print(f"done {total_time}")