uncloud-mravi/uncloud_django_based/uncloud/uncloud_pay/stripe.py
2020-04-02 19:31:03 +02:00

133 lines
4.3 KiB
Python

import stripe
import stripe.error
import logging
from django.core.exceptions import ObjectDoesNotExist
import uncloud_pay.models
import uncloud.secrets
# Static stripe configuration used below.
CURRENCY = 'chf'
stripe.api_key = uncloud.secrets.STRIPE_KEY
# Helper (decorator) used to catch errors raised by stripe logic.
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."
try:
response_object = f(*args, **kwargs)
response = {
'response_object': response_object,
'error': None
}
return response
except stripe.error.CardError as e:
# Since it's a decline, stripe.error.CardError will be caught
body = e.json_body
err = body['error']
response.update({'error': err['message']})
logging.error(str(e))
return response
except stripe.error.RateLimitError:
response.update(
{'error': "Too many requests made to the API too quickly"})
return response
except stripe.error.InvalidRequestError as e:
logging.error(str(e))
response.update({'error': "Invalid parameters"})
return response
except stripe.error.AuthenticationError as e:
# Authentication with Stripe's API failed
# (maybe you changed API keys recently)
logging.error(str(e))
response.update({'error': common_message})
return response
except stripe.error.APIConnectionError as e:
logging.error(str(e))
response.update({'error': common_message})
return response
except stripe.error.StripeError as e:
# maybe send email
logging.error(str(e))
response.update({'error': common_message})
return response
except Exception as e:
# maybe send email
logging.error(str(e))
response.update({'error': common_message})
return response
return handle_problems
# Convenience CC container, also used for serialization.
class CreditCard():
number = None
exp_year = None
exp_month = None
cvc = None
def __init__(self, number, exp_month, exp_year, cvc):
self.number=number
self.exp_year = exp_year
self.exp_month = exp_month
self.cvc = cvc
# Actual Stripe logic.
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.
customer_request = create_customer(user.username, user.email)
if customer_request['error'] == None:
mapping = uncloud_pay.models.StripeCustomer.objects.create(
owner=user,
stripe_id=customer_request['response_object']['id']
)
return mapping.stripe_id
else:
return None
@handle_stripe_error
def create_card(customer_id, credit_card):
return stripe.Customer.create_source(
customer_id,
card={
'number': credit_card.number,
'exp_month': credit_card.exp_month,
'exp_year': credit_card.exp_year,
'cvc': credit_card.cvc
})
@handle_stripe_error
def get_card(customer_id, card_id):
return stripe.Customer.retrieve_source(customer_id, card_id)
@handle_stripe_error
def charge_customer(amount, customer_id, card_id):
# Amount is in CHF but stripes requires smallest possible unit.
# See https://stripe.com/docs/api/charges/create
adjusted_amount = int(amount * 100)
return stripe.Charge.create(
amount=adjusted_amount,
currency=CURRENCY,
customer=customer_id,
source=card_id)
@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)