diff --git a/localstripe/errors.py b/localstripe/errors.py index d65327a..e7e8a09 100644 --- a/localstripe/errors.py +++ b/localstripe/errors.py @@ -26,7 +26,8 @@ def json_response(*args, **kwargs): class UserError(Exception): - def __init__(self, code, message=None, contents={}, errtype='invalid_request_error'): + def __init__(self, code, message=None, contents={}, + errtype='invalid_request_error'): Exception.__init__(self, message) self.code = code self.body = {'error': contents} diff --git a/localstripe/resources.py b/localstripe/resources.py index 5aa68c9..82cce29 100644 --- a/localstripe/resources.py +++ b/localstripe/resources.py @@ -452,7 +452,6 @@ def __init__(self, source=None, customer=None, **kwargs): metadata=self.metadata ) - @property def last4(self): return self._card_number[-4:] @@ -511,7 +510,8 @@ def __init__(self, amount=None, currency=None, description=None, if source is not None: assert type(source) is str assert (source.startswith('pm_') or source.startswith('src_') - or source.startswith('card_') or source.startswith('tok_')) + or source.startswith('card_') + or source.startswith('tok_')) assert type(capture) is bool if receipt_email is not None: assert type(receipt_email) is str and receipt_email @@ -653,19 +653,6 @@ def on_success(): obj._trigger_payment(on_success) return obj - @classmethod - def _api_list_all(cls, url, customer=None, limit=None): - try: - if customer is not None: - assert type(customer) is str and customer.startswith('cus_') - except AssertionError: - raise UserError(400, 'Bad request') - - li = super(Charge, cls)._api_list_all(url, limit=limit) - if customer is not None: - li._list = [c for c in li._list if c.customer == customer] - return li - @property def paid(self): return self.status == 'succeeded' @@ -684,7 +671,7 @@ def refunded(self): return self.amount <= self.amount_refunded @classmethod - def _api_list_all(cls, url, customer=None, created=None, limit=10, + def _api_list_all(cls, url, customer=None, created=None, limit=None, starting_after=None): try: if customer is not None: @@ -761,9 +748,11 @@ def __init__(self, id=None, name=None, duration=None, amount_off=None, f'percent_off expected to be of type float or int, got: ' \ f'{percent_off}' assert 0 <= percent_off <= 100, \ - f'percent_off must be between 0 and 100, got: {percent_off}' + ('percent_off must be between 0 and 100, ' + f'got: {percent_off}') assert duration in ('forever', 'once', 'repeating'), \ - f'duration must be one of [forever, once, repeating], got: {duration}' + ('duration must be one of [forever, once, repeating], ' + f'got:{duration}') if amount_off is not None: assert type(currency) is str and currency, \ f'currency is expected, got: {currency}' @@ -784,7 +773,7 @@ def __init__(self, id=None, name=None, duration=None, amount_off=None, if redeem_by is not None: assert type(redeem_by) is int, \ f'redeem_by expected to be an int, got: {type(redeem_by)}' - assert redeem_by > 0,\ + assert redeem_by > 0, \ f'redeem_by greater than 0 expected, got: {redeem_by}' except AssertionError: raise UserError(400, 'Bad request') @@ -805,6 +794,7 @@ def __init__(self, id=None, name=None, duration=None, amount_off=None, self.valid = True self.applies_to = applies_to + class Customer(StripeObject): object = 'customer' _id_prefix = 'cus_' @@ -913,20 +903,6 @@ def subscriptions(self): return Subscription._api_list_all( '/v1/customers/' + self.id + '/subscriptions', customer=self.id) - @classmethod - def _api_list_all(cls, url, email=None, limit=None): - try: - if email is not None: - # minimal email validation - assert type(email) is str and email.index('@') > 0 - except AssertionError: - raise UserError(400, 'Bad request') - - li = super(Customer, cls)._api_list_all(url, limit=limit) - if email is not None: - li._list = [c for c in li._list if c.email == email] - return li - @classmethod def _api_create(cls, source=None, **data): obj = super()._api_create(**data) @@ -1011,7 +987,8 @@ def _api_add_source(cls, id, source=None, **kwargs): if source_obj._attaching_is_declined(): raise UserError(402, 'Your card was declined.', - {'code': 'card_declined', 'decline_code': source_obj._decline_code()}) + {'code': 'card_declined', + 'decline_code': source_obj._decline_code()}) if isinstance(source_obj, Card): source_obj._set_customer(id) @@ -2285,7 +2262,6 @@ def __init__(self, id=None, customer=None, type=None, 'mandate_url': 'https://fake/NXDSYREGC9PSMKWY', } - def _requires_authentication(self): if self.type == 'card': return self._card_number in ('4000002500003155', @@ -2661,17 +2637,25 @@ def __init__(self, id=None, metadata=None, product=None, type(t) is dict and 'up_to' in t and \ (t['up_to'] == 'inf' or type(try_convert_to_int(t['up_to'])) is int) - unit_amount_tiers = try_convert_to_int(t.get('unit_amount', 0)) - assert type(unit_amount_tiers) is int and unit_amount_tiers >= 0 + unit_amount_tiers = try_convert_to_int(t.get('unit_amount', + 0)) + + assert (type(unit_amount_tiers) is int + and unit_amount_tiers >= 0) t['unit_amount'] = unit_amount_tiers - flat_amount_tiers = try_convert_to_int(t.get('flat_amount', 0)) - assert type(flat_amount_tiers) is int and flat_amount_tiers >= 0 + flat_amount_tiers = try_convert_to_int(t.get('flat_amount', + 0)) + + assert (type(flat_amount_tiers) is int + and flat_amount_tiers >= 0) t['flat_amount'] = flat_amount_tiers assert type(currency) is str and currency if recurring: - assert type(recurring) is dict and 'interval' in recurring and \ - 'interval_count' in recurring and \ - recurring['interval'] in ('day', 'week', 'month', 'year') + assert (type(recurring) is dict + and 'interval' in recurring + and 'interval_count' in recurring + and recurring['interval'] + in ('day', 'week', 'month', 'year')) interval_count = try_convert_to_int(recurring.get( 'interval_count', 0)) assert type(interval_count) is int and interval_count >= 0 @@ -2715,7 +2699,12 @@ def statement_descriptor(self): # Support Stripe API <= 2018-02-05 return Product._api_retrieve(self.product).statement_descriptor @classmethod - def _api_list_all(cls, url, active=True, product=None, limit=None, **kwargs): + def _api_list_all(cls, + url, + active=True, + product=None, + limit=None, + **kwargs): if kwargs: raise UserError(400, 'Unexpected ' + ', '.join(kwargs.keys())) @@ -2723,9 +2712,9 @@ def _api_list_all(cls, url, active=True, product=None, limit=None, **kwargs): li._list = [value for key, value in store.items() if key.startswith(cls.object + ':')] if active: - li._list = list(filter(lambda x: x.active == True, li._list)) + li._list = list(filter(lambda x: x.active is True, li._list)) if product: - li._list = list(filter(lambda x: x.product == product,li._list)) + li._list = list(filter(lambda x: x.product == product, li._list)) return li diff --git a/localstripe/seed_data.py b/localstripe/seed_data.py index c3e6903..baa2586 100644 --- a/localstripe/seed_data.py +++ b/localstripe/seed_data.py @@ -1,28 +1,37 @@ +from aiohttp.web_runner import GracefulExit +from glob import glob + from .errors import UserError from .resources import Product, Plan, Price, Coupon -from aiohttp.web_runner import GracefulExit -from glob import glob import json import os.path import logging + async def seed_if_does_not_exist(cls, data, logger): if not isinstance(data, list): - raise UserError(400, 'incorrect format (data provided for %s must be a list)' % cls.object) + raise UserError(400, + ('incorrect format' + '(data provided for %s must be a list)' % cls.object)) for datum in data: try: cls._api_create(**datum)._export() - logger.info("Successfully created %s: %s" %(cls.object, datum.get('id'))) + logger.info((f'Successfully created {cls.object}' + f": {datum.get('id')}")) except UserError as e: if (e.code == 409): - logger.info("Ignoring already seeded %s: %s" %(cls.object, datum.get('id'))) + logger.info(('Ignoring already seeded {cls.object}: ' + f"{datum.get('id')}")) pass else: - logger.error("Error seeding %s (%s): %s" %(cls.object, datum.get('id'), e.body)) + logger.error((f'Error seeding {cls.object} ' + f"({datum.get('id')}): " + f'{e.body}')) except Exception as e: - logger.error("Incorrect format for %s. Skipping." % cls.object) + logger.error(f'Incorrect format for {cls.object}: {e}. Skipping.') + async def seed_data(app): logger = logging.getLogger('aiohttp.access') @@ -33,7 +42,9 @@ async def seed_data(app): file_list = glob(seed_path_glob) if not file_list: - logger.warn("\n\n!!! WARNING: No fixture file found in directory: %s/ !!!", seed_dir) + logger.warn(('\n\n!!! WARNING:' + f'No fixture file found in directory: {seed_dir}/' + '!!!')) return logger.warn("Data store will not be seeded.\n") for path in file_list: @@ -45,13 +56,17 @@ async def seed_data(app): for key in data: match key: case 'products': - await seed_if_does_not_exist(Product, data[key], logger) + await seed_if_does_not_exist(Product, data[key], + logger) case 'plans': - await seed_if_does_not_exist(Plan, data[key], logger) + await seed_if_does_not_exist(Plan, data[key], + logger) case 'prices': - await seed_if_does_not_exist(Price, data[key], logger) + await seed_if_does_not_exist(Price, data[key], + logger) case 'coupons': - await seed_if_does_not_exist(Coupon, data[key], logger) + await seed_if_does_not_exist(Coupon, data[key], + logger) case _: logger.error("Unimplemented Error: %s", key) logger.info("\n...Success!\n") @@ -59,5 +74,6 @@ async def seed_data(app): logger.error("Seed Failed!!! Error decoding JSON seed file: %s\n", e) raise GracefulExit() except Exception as e: - logger.error("Seed Failed!!! Error seeding data store with data file: %s\n", e) + logger.error(('Seed Failed!!!' + f'Error seeding data store with data file: {e}\n')) raise GracefulExit() diff --git a/localstripe/server.py b/localstripe/server.py index 56e00d2..d7f0968 100644 --- a/localstripe/server.py +++ b/localstripe/server.py @@ -27,7 +27,6 @@ Invoice, InvoiceItem, PaymentIntent, PaymentMethod, Payout, Plan, \ Price, Product, Refund, SetupIntent, Source, Subscription, \ SubscriptionItem, TaxRate, Token, extra_apis, store - from .errors import UserError from .test_tokens import create_test_tokens from .seed_data import seed_data @@ -208,6 +207,7 @@ async def save_store_middleware(request, handler): save_store_middleware]) app.on_response_prepare.append(add_cors_headers) + def api_create(cls, url): async def f(request): data = await get_post_data(request) @@ -278,9 +278,9 @@ async def f(request): for cls in (BalanceTransaction, Charge, Coupon, Customer, Event, Invoice, - InvoiceItem, PaymentIntent, PaymentMethod, Payout, Plan, Price, Product, - Refund, SetupIntent, Source, Subscription, SubscriptionItem, - TaxRate, Token): + InvoiceItem, PaymentIntent, PaymentMethod, Payout, Plan, Price, + Product, Refund, SetupIntent, Source, Subscription, + SubscriptionItem, TaxRate, Token): for method, url, func in ( ('POST', '/v1/' + cls.object + 's', api_create), ('GET', '/v1/' + cls.object + 's/{id}', api_retrieve), @@ -299,9 +299,11 @@ def localstripe_js(request): app.router.add_get('/js.stripe.com/v3/', localstripe_js) + async def healthcheck(request): return web.json_response('pong') + async def config_webhook(request): id = request.match_info['id'] data = await get_post_data(request) or {} @@ -328,6 +330,7 @@ async def flush_store(request): app.on_startup.append(seed_data) + def start(): parser = argparse.ArgumentParser() parser.add_argument('--port', type=int, default=8420) @@ -351,5 +354,6 @@ def start(): create_test_tokens() web.run_app(app, sock=sock, access_log=logger) + if __name__ == '__main__': start()