capsul-flask/capsulflask/payment.py

182 lines
6.1 KiB
Python

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("payment", __name__, url_prefix="/payment")
def validate_dollars():
errors = list()
dollars = None
if "dollars" not in request.form:
errors.append("dollars is required")
else:
dollars = None
try:
dollars = decimal.Decimal(request.form["dollars"])
except:
errors.append("dollars must be a number")
# TODO re enable this
# if dollars and dollars < decimal.Decimal(1):
# errors.append("dollars must be >= 1")
return [errors, dollars]
@bp.route("/btcpay", methods=("GET", "POST"))
@account_required
def btcpay_payment():
errors = list()
invoice_id = None
if request.method == "POST":
result = validate_dollars()
errors = result[0]
dollars = result[1]
if len(errors) == 0:
invoice = current_app.config['BTCPAY_CLIENT'].create_invoice(dict(
price=float(dollars),
currency="USD",
itemDesc="Capsul Cloud Compute",
transactionSpeed="high",
redirectURL=f"{current_app.config['BASE_URL']}/account-balance"
))
# print(invoice)
invoice_id = invoice["id"]
print(f"created btcpay invoice_id={invoice_id} ( {session['account']}, ${request.form['dollars']} )")
get_model().create_payment_session("btcpay", invoice_id, session["account"], dollars)
return redirect(invoice["url"])
for error in errors:
flash(error)
return render_template("btcpay.html", invoice_id=invoice_id)
@bp.route("/stripe", methods=("GET", "POST"))
@account_required
def stripe_payment():
stripe_checkout_session_id=None
errors = list()
if request.method == "POST":
result = validate_dollars()
errors = result[0]
dollars = result[1]
if len(errors) == 0:
print(f"creating stripe checkout session for {session['account']}, ${dollars}")
checkout_session = stripe.checkout.Session.create(
success_url=current_app.config['BASE_URL'] + "/payment/stripe/success?session_id={CHECKOUT_SESSION_ID}",
cancel_url=current_app.config['BASE_URL'] + "/payment/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']}, ${dollars} )")
get_model().create_payment_session("stripe", stripe_checkout_session_id, session["account"], 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("/stripe/success", methods=("GET",))
def success():
stripe_checkout_session_id = request.args.get('session_id')
if not stripe_checkout_session_id:
print("/payment/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_payment_session deletes the checkout session row and inserts a payment row
# its ok to call consume_payment_session more than once because it only takes an action if the session exists
success_account = get_model().consume_payment_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_payment_session deletes the checkout session row and inserts a payment row
# # its ok to call consume_payment_session more than once because it only takes an action if the session exists
# success_account = get_model().consume_payment_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("/payment/stripe/webhook returned 400: bad request", e)
# abort(400, "bad request")
# except stripe.error.SignatureVerificationError:
# print("/payment/stripe/webhook returned 400: invalid signature")
# abort(400, "invalid signature")