Last active
April 17, 2021 18:56
-
-
Save freelancing-solutions/b65a7e9aa0377c97bacc8882acc2f78b to your computer and use it in GitHub Desktop.
How to create a modular flask app to support Error testing and easier code maintanability
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
import typing | |
from google.cloud import ndb | |
from google.cloud.ndb.exceptions import BadArgumentError, BadQueryError, BadRequestError, BadValueError | |
from flask import current_app, jsonify | |
from datetime import datetime | |
from data_service.store.memberships import Memberships, MembershipPlans, AccessRights | |
from data_service.store.memberships import PlanValidators as PlanValid | |
from data_service.store.users import UserValidators as UserValid | |
from data_service.store.memberships import MembershipValidators as MemberValid | |
from data_service.utils.utils import create_id | |
class Validators(UserValid, PlanValid, MemberValid): | |
def __init__(self): | |
pass | |
def can_add_member(self, uid: str, plan_id: str, start_date: datetime) -> bool: | |
user_valid: bool = self.is_user_valid(uid=uid) | |
plan_exist: bool = self.plan_exist(plan_id=plan_id) | |
date_valid: bool = self.start_date_valid(start_date=start_date) | |
return user_valid and plan_exist and date_valid | |
class MembershipsView(Validators): | |
def __init__(self): | |
super(MembershipsView, self).__init__() | |
self.client = ndb.Client(namespace="main", project=current_app.config.get('PROJECT')) | |
def _create_or_update_membership(self, uid: str, plan_id: str, plan_start_date: datetime) -> tuple: | |
with self.client.context(): | |
if self.can_add_member(uid=uid, plan_id=plan_id, start_date=plan_start_date) is True: | |
try: | |
member_ships_list: typing.List[Memberships] = Memberships.query(Memberships.uid == uid).fetch() | |
if isinstance(member_ships_list, list) and len(member_ships_list) > 0: | |
membership_instance: Memberships = member_ships_list[0] | |
else: | |
membership_instance: Memberships = Memberships() | |
membership_instance.uid = uid | |
membership_instance.plan_id = create_id() | |
membership_instance.status = 'Unpaid' | |
membership_instance.date_created = datetime.now() | |
membership_instance.plan_start_date = plan_start_date | |
key = membership_instance.put() | |
if key is None: | |
return jsonify({'status': False, 'message': 'for some reason we where unable to create '}), 500 | |
except ValueError as e: | |
message: str = str(e) | |
return jsonify({'status': False, 'message': message}), 500 | |
except TypeError as e: | |
message: str = str(e) | |
return jsonify({'status': False, 'message': message}), 500 | |
except BadRequestError as e: | |
message: str = str(e) | |
return jsonify({'status': False, 'message': message}), 500 | |
except BadQueryError as e: | |
message: str = str(e) | |
return jsonify({'status': False, 'message': message}), 500 | |
return jsonify({'status': True, 'message': 'successfully updated membership', | |
'payload': membership_instance.to_dict()}), 200 | |
def add_membership(self, uid: str, plan_id: str, plan_start_date: datetime) -> tuple: | |
return self._create_or_update_membership(uid=uid, plan_id=plan_id, plan_start_date=plan_start_date) | |
def update_membership(self, uid: str, plan_id: str, plan_start_date: datetime) -> tuple: | |
return self._create_or_update_membership(uid=uid, plan_id=plan_id, plan_start_date=plan_start_date) | |
def set_membership_status(self, uid: str, status: str) -> tuple: | |
with self.client.context(): | |
membership_list: typing.List[Memberships] = Memberships.query(Memberships.uid == uid).fetch() | |
if isinstance(membership_list, list) and len(membership_list) > 0: | |
membership_instance: Memberships = membership_list[0] | |
try: | |
membership_instance.status = status | |
key = membership_instance.put() | |
if key is None: | |
message: str = 'For some reason we where unable to update the membership status' | |
return jsonify({'status': False, 'message': message}), 500 | |
except ValueError as e: | |
message: str = str(e) | |
return jsonify({'status': False, 'message': message}), 500 | |
except BadRequestError as e: | |
message: str = str(e) | |
return jsonify({'status': False, 'message': message}), 500 | |
except BadQueryError as e: | |
message: str = str(e) | |
return jsonify({'status': False, 'message': message}), 500 | |
message: str = "Successfully update membership status" | |
return jsonify({'status': True, 'payload': membership_instance.to_dict(), 'message': message}) | |
def change_membership(self, uid: str, origin_plan_id: str, dest_plan_id: str) -> tuple: | |
with self.client.context(): | |
try: | |
membership_list: typing.List[Memberships] = Memberships.query(Memberships.uid == uid).fetch() | |
if isinstance(membership_list, list) and len(membership_list) > 0: | |
membership_instance: Memberships = membership_list[0] | |
if membership_instance.plan_id == origin_plan_id: | |
if self.plan_exist(plan_id=dest_plan_id): | |
membership_instance.plan_id = dest_plan_id | |
key = membership_instance.put(use_cache=True, retries=5) | |
if key is None: | |
message: str = "for some reason we are unable to change the membership" | |
return jsonify({'status': False, 'message': message}), 200 | |
except ValueError as e: | |
message: str = str(e) | |
return jsonify({'status': False, 'message': message}), 500 | |
except TypeError as e: | |
message: str = str(e) | |
return jsonify({'status': False, 'message': message}), 500 | |
except BadRequestError as e: | |
message: str = str(e) | |
return jsonify({'status': False, 'message': message}), 500 | |
except BadQueryError as e: | |
message: str = str(e) | |
return jsonify({'status': False, 'message': message}), 500 | |
return jsonify({'status': True, 'message': 'successfully updated membership', | |
'payload': membership_instance.to_dict()}), 200 | |
def send_welcome_email(self, uid: str, plan_id: str) -> tuple: | |
""" | |
just send a request to the email service to send emails | |
""" | |
with self.client.context(): | |
pass | |
def return_plan_members_by_payment_status(self, plan_id: str, status: bool) -> tuple: | |
""" | |
for members of this plan_id return members by payment_status | |
""" | |
with self.client.context(): | |
membership_list: typing.List[Memberships] = Memberships.query( | |
Memberships.plan_id == plan_id, Memberships.status == status).fetch() | |
response_data: typing.List[dict] = [member.to_dict() for member in membership_list] | |
return jsonify({'status': True, 'payload': response_data, 'message': 'successfully fetched members'}), 200 | |
def return_plan_members(self, plan_id) -> tuple: | |
""" | |
return all members of a plan | |
""" | |
with self.client.context(): | |
membership_list: typing.List[Memberships] = Memberships.query(Memberships.plan_id == plan_id).fetch() | |
response_data: typing.List[dict] = [member.to_dict() for member in membership_list] | |
return jsonify({'status': True, 'payload': response_data, 'message': 'successfully fetched members'}), 200 | |
def is_member_off(self, uid: str) -> tuple: | |
""" | |
returns user membership details | |
""" | |
with self.client.context(): | |
membership_list: typing.List[Memberships] = Memberships.query(Memberships.uid == uid).fetch() | |
response_data: typing.List[dict] = [member.to_dict() for member in membership_list] | |
if isinstance(response_data, list) and len(response_data): | |
return jsonify( | |
{'status': True, 'payload': response_data[0], 'message': 'successfully fetched members'}), 200 | |
else: | |
return jsonify({'status': False, 'message': 'user does not have any membership plan'}), 500 | |
def payment_amount(self, uid: str) -> tuple: | |
""" | |
for a specific user return payment amount | |
""" | |
with self.client.context(): | |
membership_list: typing.List[Memberships] = Memberships.query(Memberships.uid == uid).fetch() | |
if isinstance(membership_list, list) and len(membership_list) > 0: | |
membership_instance: Memberships = membership_list[0] | |
plan_id = membership_instance.plan_id | |
membership_plan_instance: MembershipPlans = MembershipPlansView().get_plan(plan_id=plan_id) | |
if membership_plan_instance is None: | |
return jsonify({'status': False, 'message': 'could not find plan associate with the plan_id'}), 500 | |
amount_data: dict = { | |
'term_payment_amount': membership_plan_instance.term_payment_amount, | |
'registration_amount': membership_plan_instance.registration_amount} | |
message: str = 'successfully returned payment details' | |
return jsonify({'status': True, 'payload': amount_data, 'message': message}), 200 | |
message: str = 'unable to locate membership details' | |
return jsonify({'status': False, 'message': message}), 500 | |
def set_payment_status(self, uid: str, status: str) -> tuple: # status is paid or unpaid | |
""" | |
for a specific user set payment status | |
""" | |
with self.client.context(): | |
try: | |
membership_list: typing.List[Memberships] = Memberships.query(Memberships.uid == uid).fetch() | |
if isinstance(membership_list, list) and len(membership_list) > 0: | |
membership_instance: Memberships = membership_list[0] | |
membership_instance.status = status | |
key = membership_instance.put() | |
if key is None: | |
message: str = 'for some reason we are unable to set payment status' | |
return jsonify({'status': False, 'message': message}), 500 | |
except ValueError as e: | |
message: str = str(e) | |
return jsonify({'status': False, 'message': message}), 500 | |
except TypeError as e: | |
message: str = str(e) | |
return jsonify({'status': False, 'message': message}), 500 | |
except BadRequestError as e: | |
message: str = str(e) | |
return jsonify({'status': False, 'message': message}), 500 | |
except BadQueryError as e: | |
message: str = str(e) | |
return jsonify({'status': False, 'message': message}), 500 | |
return jsonify({'status': True, 'message': 'payment status has been successfully set', | |
'payload': membership_instance.to_dict()}), 200 | |
class MembershipPlansView: | |
def __init__(self): | |
self.client = ndb.Client(namespace="main", project=current_app.config.get('PROJECT')) | |
def add_plan(self, membership_name: str, description: str, schedule_day: int, schedule_term: str) -> tuple: | |
pass | |
def update_plan(self, plan_id, membership_name: str, description: str, schedule_day: int, | |
schedule_term: str) -> tuple: | |
pass | |
def set_plan_status(self, plan_id: str, status: bool) -> tuple: | |
pass | |
def return_monthly_plans(self) -> tuple: | |
pass | |
def return_quarterly_plan(self) -> tuple: | |
pass | |
def return_annual_plans(self) -> tuple: | |
pass | |
def get_plan(self, plan_id: str) -> typing.Union[None, MembershipPlans]: | |
with self.client.context(): | |
if isinstance(plan_id, str): | |
membership_plans_list: typing.List[MembershipPlans] = MembershipPlans.query(MembershipPlans.plan_id == plan_id).fetch() | |
if isinstance(membership_plans_list, list) and len(membership_plans_list) > 0: | |
membership_instance: MembershipPlans = membership_plans_list[0] | |
return membership_instance | |
else: | |
return None | |
return None | |
class AccessRightsView: | |
def __init__(self): | |
self.client = ndb.Client(namespace="main", project=current_app.config.get('PROJECT')) | |
def get_access_rights(self, plan_id: str) -> typing.Union[None, AccessRights]: | |
with self.client.context(): | |
if isinstance(plan_id, str): | |
access_rights_list: typing.List[AccessRights] = AccessRights.query(AccessRights.plan_id == plan_id) | |
if isinstance(access_rights_list, list) and len(access_rights_list) > 0: | |
access_rights_instance: AccessRights = access_rights_list[0] | |
return access_rights_instance | |
return None | |
return None |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Modular Python Flask App
There are several reason as to why one should write modular apps, those reasons all have to do
with Code Quality, Maintainability, Testing, Modules can be easily assigned to teams,
Easier to refactor into Micro Services.
To cut the story short modular python apps scale better.
For my freelance profile website i choose to modularization my code base by making use of blueprints and several
flask modules such as current_app.
This has the advantage that i can easily deploy the different modules in different docker containers should
the need arise, and it will.
Basically each module is registered
like this
See the Gist above for reference.