flask_stripe_checkout_subscription_with_billing_portal_webhook.py

1import stripe
2from flask import Flask, redirect, request
3
4# This is your test secret API key.
5stripe.api_key = "sk_test_51..."
6
7app = Flask(__name__)
8
9YOUR_DOMAIN = 'http://localhost:4242'
10
11@app.route('/create-checkout-session', methods=['POST'])
12def create_checkout_session():
13    try:
14        prices = stripe.Price.list(
15            lookup_keys=[request.form.get('lookup_key')],
16            expand=['data.product']
17        )
18
19        checkout_session = stripe.checkout.Session.create(
20            line_items=[
21                {
22                    'price': prices.data[0].id,
23                    'quantity': 1,
24                },
25            ],
26            mode='subscription',
27            success_url=YOUR_DOMAIN + '/success.html?session_id={CHECKOUT_SESSION_ID}',
28            cancel_url=YOUR_DOMAIN + '/cancel.html',
29        )
30        return redirect(checkout_session.url, code=303)
31    except Exception as e:
32        print(e)
33        return "Server error", 500
34
35@app.route('/create-portal-session', methods=['POST'])
36def customer_portal():
37    # For demonstration purposes, we're using the checkout session to retrieve the customer ID.
38    # Typically this is stored alongside the authenticated user in your database.
39    checkout_session_id = request.form.get('session_id')
40    checkout_session = stripe.checkout.Session.retrieve(checkout_session_id)
41
42    # This is the URL to which the customer will be redirected after they are
43    # done managing their billing with the portal.
44    return_url = YOUR_DOMAIN
45
46    portalSession = stripe.billing_portal.Session.create(
47        customer=checkout_session.customer,
48        return_url=return_url,
49    )
50    return redirect(portalSession.url, code=303)
51
52@app.route('/webhook', methods=['POST'])
53def webhook_received():
54    # Replace this endpoint secret with your endpoint's unique secret
55    # If you are testing with the Stripe CLI, run the following command:
56    # stripe listen --forward-to localhost:4242/webhook
57    webhook_secret = 'whsec_12345'
58    request_data = request.data
59    payload = request_data
60    sig_header = request.headers.get('stripe-signature')
61
62    if webhook_secret:
63        try:
64            event = stripe.Webhook.construct_event(
65                payload, sig_header, webhook_secret
66            )
67            data = event['data']
68        except Exception as e:
69            return e
70        event_type = event['type']
71    else:
72        data = request_data['data']
73        event_type = request_data['type']
74    data_object = data['object']
75
76    print('event ' + event_type)
77
78    if event_type == 'checkout.session.completed':
79        print('Payment succeeded!')
80    elif event_type == 'customer.subscription.deleted':
81        # handle subscription cancelled
82        print('Subscription cancelled')
83
84    return {'status': 'success'}
85
86if __name__ == '__main__':
87    app.run(port=4242)