uncloud-mravi/uncloud_pay/stripe.py

import stripe
import stripe.error
import logging

from django.core.exceptions import ObjectDoesNotExist
from django.conf import settings

import uncloud_pay.models

CURRENCY = 'chf'

stripe.api_key = settings.STRIPE_KEY

def handle_stripe_error(f):
    def handle_problems(*args, **kwargs):
        response = {
            'paid': False,
            'response_object': None,
            'error': None
        }

        common_message = "Currently it is not possible to make payments. Please try agin later."
        try:
            response_object = f(*args, **kwargs)
            return response_object
        except stripe.error.CardError as e:
            # Since it's a decline, stripe.error.CardError will be caught
            body = e.json_body
            logging.error(str(e))

            raise e # For error handling.
        except stripe.error.RateLimitError:
            logging.error("Too many requests made to the API too quickly.")
            raise Exception(common_message)
        except stripe.error.InvalidRequestError as e:
            logging.error(str(e))
            raise Exception('Invalid parameters.')
        except stripe.error.AuthenticationError as e:
            # Authentication with Stripe's API failed
            # (maybe you changed API keys recently)
            logging.error(str(e))
            raise Exception(common_message)
        except stripe.error.APIConnectionError as e:
            logging.error(str(e))
            raise Exception(common_message)
        except stripe.error.StripeError as e:
            # XXX: maybe send email
            logging.error(str(e))
            raise Exception(common_message)
        except Exception as e:
            # maybe send email
            logging.error(str(e))
            raise Exception(common_message)

    return handle_problems

def public_api_key():
    return settings.STRIPE_PUBLIC_KEY

def get_customer_id_for(user):
    try:
        # .get() raise if there is no matching entry.
        return uncloud_pay.models.StripeCustomer.objects.get(owner=user).stripe_id
    except ObjectDoesNotExist:
        # No entry yet - making a new one.
        try:
            customer = create_customer(user.username, user.email)
            uncloud_stripe_mapping = uncloud_pay.models.StripeCustomer.objects.create(
                    owner=user, stripe_id=customer.id)
            return uncloud_stripe_mapping.stripe_id
        except Exception as e:
            return None

@handle_stripe_error
def create_setup_intent(customer_id):
    return stripe.SetupIntent.create(customer=customer_id)

@handle_stripe_error
def get_setup_intent(setup_intent_id):
    return stripe.SetupIntent.retrieve(setup_intent_id)

def get_payment_method(payment_method_id):
    return stripe.PaymentMethod.retrieve(payment_method_id)

@handle_stripe_error
def charge_customer(amount, customer_id, card_id):
    # Amount is in CHF but stripes requires smallest possible unit.
    # https://stripe.com/docs/api/payment_intents/create#create_payment_intent-amount
    adjusted_amount = int(amount * 100)
    return stripe.PaymentIntent.create(
        amount=adjusted_amount,
        currency=CURRENCY,
        customer=customer_id,
        payment_method=card_id,
        off_session=True,
        confirm=True,
    )

@handle_stripe_error
def create_customer(name, email):
    return stripe.Customer.create(name=name, email=email)

@handle_stripe_error
def get_customer(customer_id):
    return stripe.Customer.retrieve(customer_id)