Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def _api_void_invoice(cls, id):
obj = Invoice._api_retrieve(id)
if obj.status not in ('draft', 'open'):
raise UserError(400, 'Bad request')
PaymentIntent._api_cancel(obj.payment_intent)
obj._draft = False
obj._voided = True
obj.status_transitions['voided_at'] = int(time.time())
if obj.subscription:
sub = Subscription._api_retrieve(obj.subscription)
sub._on_initial_payment_voided(obj)
return obj
def _api_add_tax_id(cls, id, type=None, value=None, **kwargs):
if kwargs:
raise UserError(400, 'Unexpected ' + ', '.join(kwargs.keys()))
try:
assert type in ('eu_vat', 'nz_gst', 'au_abn')
assert _type(value) is str and len(value) > 10
except AssertionError:
raise UserError(400, 'Bad request')
obj = cls._api_retrieve(id)
tax_id = TaxId(customer=id, type=type, value=value)
obj.tax_ids._list.append(tax_id)
return tax_id
def _api_list_subscriptions(cls, id, **kwargs):
if kwargs:
raise UserError(400, 'Unexpected ' + ', '.join(kwargs.keys()))
return cls._api_retrieve(id).subscriptions
def _api_list_tax_ids(cls, id, **kwargs):
if kwargs:
raise UserError(400, 'Unexpected ' + ', '.join(kwargs.keys()))
obj = cls._api_retrieve(id)
return obj.tax_ids
async def error_middleware(request, handler):
try:
return await handler(request)
except UserError as e:
return e.to_response()
def _api_delete(cls, id):
key = cls.object + ':' + id
if key not in store.keys():
raise UserError(404, 'Not Found')
del store[key]
return {"deleted": True, "id": id}
**kwargs):
if kwargs:
raise UserError(400, 'Unexpected ' + ', '.join(kwargs.keys()))
success = try_convert_to_bool(success)
try:
assert type(id) is str and id.startswith('pi_')
assert type(client_secret) is str
assert type(success) is bool
except AssertionError:
raise UserError(400, 'Bad request')
obj = cls._api_retrieve(id)
if client_secret != obj.client_secret:
raise UserError(401, 'Unauthorized')
if obj.status != 'requires_action':
raise UserError(400, 'Bad request')
obj.next_action = None
if success:
obj._trigger_payment()
else:
obj._authentication_failed = True
obj.payment_method = None
if obj.invoice:
invoice = Invoice._api_retrieve(obj.invoice)
invoice._on_payment_failure_later()
return obj
async def f(request):
id = request.match_info['id']
data = await get_post_data(request)
if not data:
raise UserError(400, 'Bad request')
expand = data.pop('expand', None)
return json_response(cls._api_update(id, **data)._export(
expand=expand))
return f
def _api_retrieve(cls, id):
obj = store.get(cls.object + ':' + id, None)
if obj is None:
raise UserError(404, 'Not Found')
return obj
def _api_confirm(cls, id, use_stripe_sdk=None, client_secret=None,
payment_method_data=None, **kwargs):
if kwargs:
raise UserError(400, 'Unexpected ' + ', '.join(kwargs.keys()))
try:
assert type(id) is str and id.startswith('seti_')
if client_secret is not None:
assert type(client_secret) is str
if payment_method_data is not None:
assert type(payment_method_data) is dict
except AssertionError:
raise UserError(400, 'Bad request')
obj = cls._api_retrieve(id)
if client_secret and client_secret != obj.client_secret:
raise UserError(401, 'Unauthorized')
if payment_method_data:
if obj.payment_method is not None:
raise UserError(400, 'Bad request')
pm = PaymentMethod(**payment_method_data)
obj.payment_method = pm.id
if pm._attaching_is_declined():
obj.status = 'canceled'
obj.next_action = None