Fix a bunch of Flake8 issues

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}

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,17 +2699,22 @@ 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()))
 
         li = List(url, limit=limit)
         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
 
 

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,19 +56,24 @@ 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")
     except json.JSONDecodeError as e:
         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()

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()