import stripe import json import time import decimal from flask import Blueprint from flask import request from flask import current_app from flask import session from flask import redirect from flask import url_for from flask import jsonify from flask import flash from flask import render_template from werkzeug.exceptions import abort from capsulflask.auth import account_required from capsulflask.db import get_model bp = Blueprint("stripe", __name__, url_prefix="/stripe") @bp.route("/", methods=("GET", "POST")) @account_required def index(): stripe_checkout_session_id=None errors = list() if request.method == "POST": if "dollars" not in request.form: errors.append("dollars is required") dollars = None try: dollars = decimal.Decimal(request.form["dollars"]) except: errors.append("dollars must be a number") if dollars and dollars < decimal.Decimal(1): errors.append("dollars must be >= 1") if len(errors) == 0: print(f"creating stripe checkout session for {session['account']}, ${request.form['dollars']}") checkout_session = stripe.checkout.Session.create( success_url=current_app.config['BASE_URL'] + "/stripe/success?session_id={CHECKOUT_SESSION_ID}", cancel_url=current_app.config['BASE_URL'] + "/stripe", payment_method_types=["card"], customer_email=session["account"], line_items=[ { "name": "Capsul Cloud Compute", "images": [current_app.config['BASE_URL']+"/static/capsul-product-image.png"], "quantity": 1, "currency": "usd", "amount": int(dollars*100) } ] ) stripe_checkout_session_id = checkout_session['id'] print(f"stripe_checkout_session_id={stripe_checkout_session_id} ( {session['account']}, ${request.form['dollars']} )") get_model().create_stripe_checkout_session(stripe_checkout_session_id, session["account"], request.form["dollars"]) for error in errors: flash(error) return render_template( "stripe.html", stripe_checkout_session_id=stripe_checkout_session_id, stripe_public_key=current_app.config["STRIPE_PUBLISHABLE_KEY"] ) @bp.route("/success", methods=("GET",)) def success(): stripe_checkout_session_id = request.args.get('session_id') if not stripe_checkout_session_id: print("/stripe/success returned 400: missing required URL parameter session_id") abort(400, "missing required URL parameter session_id") else: checkout_session_completed_events = stripe.Event.list( type='checkout.session.completed', created={ # Check for events created in the last half hour 'gte': int(time.time() - (30 * 60)), }, ) for event in checkout_session_completed_events.auto_paging_iter(): checkout_session = event['data']['object'] if checkout_session and 'id' in checkout_session and checkout_session['id'] == stripe_checkout_session_id: cents = checkout_session['display_items'][0]['amount'] dollars = decimal.Decimal(cents)/100 #consume_stripe_checkout_session deletes the checkout session row and inserts a payment row # its ok to call consume_stripe_checkout_session more than once because it only takes an action if the session exists success_account = get_model().consume_stripe_checkout_session(stripe_checkout_session_id, dollars) if success_account: print(f"{success_account} paid ${dollars} successfully (stripe_checkout_session_id={stripe_checkout_session_id})") return redirect(url_for("console.account_balance")) abort(400, "this checkout session is not paid yet") # webhook is not needed # @bp.route("/webhook", methods=("POST",)) # def webhook(): # request_data = json.loads(request.data) # signature = request.headers.get('stripe-signature') # try: # event = stripe.Webhook.construct_event( # payload=request_data, # sig_header=signature, # secret=current_app.config['STRIPE_WEBHOOK_SECRET'] # ) # if event['type'] == 'checkout.session.completed': # dollars = event['data']['object']['display_items'][0]['amount'] # stripe_checkout_session_id = event['data']['object']['id'] # #consume_stripe_checkout_session deletes the checkout session row and inserts a payment row # # its ok to call consume_stripe_checkout_session more than once because it only takes an action if the session exists # success_account = get_model().consume_stripe_checkout_session(stripe_checkout_session_id, dollars) # if success_account: # print(f"{success_account} paid ${dollars} successfully (stripe_checkout_session_id={stripe_checkout_session_id})") # return jsonify({'status': 'success'}) # except ValueError as e: # print("/stripe/webhook returned 400: bad request", e) # abort(400, "bad request") # except stripe.error.SignatureVerificationError: # print("/stripe/webhook returned 400: invalid signature") # abort(400, "invalid signature")