flask_stripe_webhook_listener_with_signature_verification.py

1import json
2import os
3import stripe
4from flask import Flask, jsonify, request
5
6# This is your Stripe CLI webhook secret for testing your endpoint locally.
7endpoint_secret = 'whsec_...'
8
9stripe.api_key = 'sk_test_...'
10
11app = Flask(__name__)
12
13@app.route('/webhook', methods=['POST'])
14def webhook():
15    event = None
16    payload = request.data
17    sig_header = request.headers.get('STRIPE_SIGNATURE')
18
19    try:
20        event = stripe.Webhook.construct_event(
21            payload, sig_header, endpoint_secret
22        )
23    except ValueError as e:
24        # Invalid payload
25        return 'Invalid payload', 400
26    except stripe.error.SignatureVerificationError as e:
27        # Invalid signature
28        return 'Invalid signature', 400
29
30    # Handle the event
31    if event['type'] == 'payment_intent.succeeded':
32      payment_intent = event['data']['object']
33      print('PaymentIntent was successful!')
34    # ... handle other event types
35    else:
36      print('Unhandled event type {}'.format(event['type']))
37
38    return jsonify(success=True)
39
40if __name__ == '__main__':
41    app.run(port=4242)