Skip to content
This repository has been archived by the owner on Aug 7, 2024. It is now read-only.

Commit

Permalink
Fix a bunch of Flake8 issues
Browse files Browse the repository at this point in the history
  • Loading branch information
@calm-mlin
calm-mlin committed Jul 29, 2024
1 parent 0c03193 commit 4811d5c
Show file tree
Hide file tree
Showing 4 changed files with 75 additions and 65 deletions.
3 changes: 2 additions & 1 deletion localstripe/errors.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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}
Expand Down
79 changes: 34 additions & 45 deletions localstripe/resources.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -452,7 +452,6 @@ def __init__(self, source=None, customer=None, **kwargs):
metadata=self.metadata
)


@property
def last4(self):
return self._card_number[-4:]
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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'
Expand All @@ -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:
Expand Down Expand Up @@ -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}'
Expand All @@ -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')
Expand All @@ -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_'
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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',
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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


Expand Down
46 changes: 31 additions & 15 deletions localstripe/seed_data.py
View file
Original file line number Diff line number Diff line change
@@ -1,28 +1,37 @@
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

from aiohttp.web_runner import GracefulExit

from .errors import UserError
from .resources import Product, Plan, Price, Coupon


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')
Expand All @@ -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:
Expand All @@ -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()
12 changes: 8 additions & 4 deletions localstripe/server.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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),
Expand All @@ -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 {}
Expand All @@ -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)
Expand All @@ -351,5 +354,6 @@ def start():
create_test_tokens()
web.run_app(app, sock=sock, access_log=logger)


if __name__ == '__main__':
start()

0 comments on commit 4811d5c

Please sign in to comment.