Scott.Idem/OSIT-AE-API-FastAPI
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Working on all the routes, methods, and models.

Browse Source
This commit is contained in:
Scott Idem
2021-06-18 17:49:32 -04:00
parent 938aabb2a8
commit ce44e93b4d
17 changed files with 614 additions and 285 deletions
Download Patch File Download Diff File Expand all files Collapse all files

204
app/methods/membership_member_methods.py
View File

@@ -7,13 +7,12 @@ from pydantic import BaseModel, EmailStr, Field, PrivateAttr, ValidationError, v
from app.lib_general import log, logging
from app.db_sql import redis_lookup_id_random, sql_insert_or_update, sql_select
#from .address_models import Address_Base
#from .contact_models import Contact_Base
from app.methods.membership_group_methods import get_membership_group_rec_list, load_membership_group_obj
from app.methods.membership_profile_methods import get_membership_profile_rec_list, load_membership_profile_obj
from app.methods.person_methods import load_person_obj
from app.methods.user_methods import load_user_obj
from app.models.membership_member_models import Membership_Member_Base
from app.models.membership_profile_models import Membership_Profile_Base
#from .organization_models import Organization_Base
#from .person_models import Person_Base
#from .user_models import User_Base
# ### BEGIN ### API Membership Member Methods ### save_membership_member_obj() ###
@@ -167,6 +166,8 @@ def save_membership_member_obj(order_obj_new:Membership_Member_Base=None):
def load_membership_member_obj(
membership_member_id:int|str,
limit: int = 1000,
by_alias: bool = True,
exclude_unset: bool = True,
model_as_dict: bool = False,
enabled: str = 'enabled', # enabled, disabled, all
inc_address: bool = False,
@@ -194,196 +195,89 @@ def load_membership_member_obj(
except ValidationError as e:
log.error(e.json())
# Updated 2021-06-18
if inc_membership_group_list:
if membership_group_rec_list_result := get_membership_group_rec_list(
for_obj_type = 'membership_member',
for_obj_id = membership_member_id,
limit = limit,
enabled = enabled,
):
membership_group_result_list = []
for membership_group_rec in membership_group_rec_list_result:
membership_group_result_list.append(
load_membership_group_obj(
membership_group_id = membership_group_rec.get('membership_group_id', None),
# limit = limit,
by_alias = by_alias,
exclude_unset = exclude_unset,
model_as_dict = model_as_dict,
# enabled = enabled,
)
)
membership_member_obj.membership_group_list = membership_group_result_list
else: membership_member_obj.membership_group_list = []
# Updated 2021-06-18
if inc_membership_profile:
membership_profile_id = membership_member_rec.get('membership_profile_id', None)
log.debug(membership_profile_id)
if membership_profile_dict := load_membership_profile_obj(
if membership_profile_result := load_membership_profile_obj(
membership_profile_id = membership_profile_id,
by_alias = by_alias,
exclude_unset = exclude_unset,
model_as_dict = model_as_dict,
inc_address = inc_address,
inc_contact = inc_contact,
# inc_membership = inc_membership,
# inc_organization = inc_organization,
):
membership_member_obj.membership_profile = membership_profile_dict
membership_member_obj.membership_profile = membership_profile_result
else: membership_member_obj.membership_profile = None
log.debug(membership_profile_dict)
log.debug(membership_profile_result)
# Updated 2021-06-18
if inc_person:
person_id = membership_member_rec.get('person_id', None)
log.debug(person_id)
if person_dict := load_person_obj(
if person_result := load_person_obj(
person_id = person_id,
by_alias = by_alias,
exclude_unset = exclude_unset,
model_as_dict = model_as_dict,
inc_address = inc_address,
inc_contact = inc_contact,
# inc_organization = inc_organization,
# inc_user = inc_user,
):
membership_member_obj.person = person_dict
membership_member_obj.person = person_result
else: membership_member_obj.person = None
log.debug(person_dict)
log.debug(person_result)
# Updated 2021-06-18
if inc_user:
user_id = membership_member_rec.get('user_id', None)
if user_dict := load_user_obj(
if user_result := load_user_obj(
user_id = user_id,
by_alias = by_alias,
exclude_unset = exclude_unset,
model_as_dict = model_as_dict,
inc_address = inc_address,
inc_contact = inc_contact,
# inc_organization = inc_organization,
# inc_person = inc_person,
):
membership_member_obj.user = user_dict
membership_member_obj.user = user_result
else: membership_member_obj.user = None
log.debug(user_dict)
if membership_member_rec := sql_select(table_name='v_membership', record_id=membership_member_id):
#log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(membership_member_rec)
if inc_membership_member_profile:
if membership_member_profile_rec := sql_select(table_name='v_membership_member_profile', field_name='membership_member_id', field_value=membership_member_id):
#log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(membership_member_profile_rec)
membership_member_rec['profile'] = membership_member_profile_rec
if inc_membership_member_cfg:
if membership_member_cfg_rec := sql_select(table_name='v_membership_member_cfg', field_name='account_id', field_value=membership_member_rec.get('account_id', None)):
#log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(membership_member_cfg_rec)
membership_member_rec['cfg'] = membership_member_cfg_rec
if inc_extended_profile:
account_code = membership_member_rec.get('account_code', None)
table_name = f'c_{account_code}_membership_member_profile'
if extended_profile_rec := sql_select(table_name=table_name, field_name='membership_member_id', field_value=membership_member_id):
#log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(extended_profile_rec)
membership_member_rec['extended_profile'] = extended_profile_rec
if inc_person:
if person_rec := sql_select(table_name='v_person', record_id=membership_member_rec.get('person_id')):
#log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(person_rec)
membership_member_rec['person'] = person_rec
if inc_user:
if user_rec := sql_select(table_name='v_user', record_id=membership_member_rec.get('user_id')):
#log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(user_rec)
membership_member_rec['user'] = user_rec
#log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(membership_member_rec)
else:
return False
log.debug(user_result)
if model_as_dict:
return membership_member_obj.dict(by_alias=True, exclude_unset=True) # pylint: disable=no-member
return membership_member_obj.dict(by_alias=by_alias, exclude_unset=exclude_unset) # pylint: disable=no-member
else:
return membership_member_obj
# ### END ### API Membership Member Methods ### load_membership_member_obj() ###
# ### BEGIN ### API Membership Member Methods ### load_membership_member_obj_list() ###
def load_membership_member_obj_list(
account_id: int|str|None = None,
membership_id: int|str|None = None,
limit: int = 1000,
model_as_dict: bool = False,
enabled: str = 'enabled', # enabled, disabled, all
inc_address: bool = False,
inc_contact: bool = False,
inc_membership_cfg: bool = False,
inc_membership_group_list: bool = False,
inc_membership_profile: bool = False,
# inc_membership_profile_cust: bool = False,
inc_organization: bool = False,
inc_person: bool = False,
inc_user: bool = False
) -> list|bool:
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
data = {}
if account_id:
if account_id := redis_lookup_id_random(record_id_random=account_id, table_name='account'): pass
else: return False
data['account_id'] = account_id
sql_obj_type_id = f'`tbl`.account_id = :account_id'
elif membership_id:
if membership_id := redis_lookup_id_random(record_id_random=membership_id, table_name='membership'): pass
else: return False
data['membership_id'] = membership_id
sql_obj_type_id = f'`tbl`.membership_id = :membership_id'
if enabled in ['enabled', 'disabled', 'all']:
if enabled == 'enabled':
data['enable'] = True
sql_enabled = f'AND `tbl`.enable = :enable'
elif enabled == 'disabled':
data['enable'] = False
sql_enabled = f'AND `tbl`.enable = :enable'
elif enabled == 'all':
sql_enabled = ''
# else: tbl_obj['account'] = None
if limit:
data['limit'] = limit
sql_limit = f'LIMIT :limit'
else:
sql_limit = ''
sql = f"""
SELECT `tbl`.id AS 'membership_member_id', `tbl`.id_random AS 'membership_member_id_random'
FROM `membership_member` AS `tbl`
WHERE
{sql_obj_type_id}
{sql_enabled}
ORDER BY `tbl`.created_on DESC, `tbl`.updated_on DESC
{sql_limit};
"""
if membership_member_rec_li_result := sql_select(data=data, sql=sql, as_list=True):
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(membership_member_rec_li_result)
membership_member_result_li = []
for membership_member_rec in membership_member_rec_li_result:
membership_member_id = membership_member_rec.get('membership_member_id', None)
if membership_member_result := load_membership_member_obj(
membership_member_id = membership_member_id,
model_as_dict = model_as_dict,
enabled = enabled,
# inc_membership = inc_membership,
inc_membership_group_list = inc_membership_group_list,
inc_membership_profile = inc_membership_profile,
inc_organization = inc_organization,
inc_person = inc_person,
# inc_product = inc_product,
inc_user = inc_user,
):
log.debug(membership_member_result)
membership_member_result_li.append(membership_member_result)
else:
log.debug(membership_member_result)
membership_member_result_li.append(None)
log.debug(membership_member_result_li)
else:
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(membership_member_rec_li_result)
membership_member_result_li = []
return membership_member_result_li
# ### END ### API Membership Member Methods ### load_membership_member_obj_list() ###
# ### BEGIN ### API Membership Member Methods ### get_membership_member_rec_list() ###
def get_membership_member_rec_list(
for_obj_type: str,