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

A LOT of code clean up. More changes to membership tables, views, routes, methods, and models renamed.

Browse Source
This commit is contained in:
Scott Idem
2022-01-11 19:45:55 -05:00
parent 6e764b21e1
commit 55063d0f94
8 changed files with 328 additions and 186 deletions
Download Patch File Download Diff File Expand all files Collapse all files

209
app/methods/membership_person_methods.py
View File

@@ -386,16 +386,17 @@ def get_membership_person_rec_list(
# ### BEGIN ### API Membership Person Methods ### create_membership_person_obj() ###
# Updated 2022-01-04
def create_membership_person_obj(
account_id: int|str,
person_id: int|str,
membership_person_dict_obj: Membership_Person_Base|dict,
account_id: int,
person_id: int,
membership_person_dict_obj: Membership_Person_Base,
create_sub_obj: bool = False,
fail_any: bool = False, # Fail if any thing goes wrong for sub objects
return_outline: bool = False,
) -> int|bool:
) -> bool|dict|int:
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
# ### SECTION ### Secondary data validation
log.info('Checking requirements...')
if account_id := redis_lookup_id_random(record_id_random=account_id, table_name='account'): pass
else:
@@ -406,27 +407,39 @@ def create_membership_person_obj(
log.error('Missing or invalid Person ID passed. Failed requirement.')
log.info(f'Person ID: {person_id}')
log.info('Create dictionary or Pydantic object')
log.debug(type(membership_person_dict_obj))
if isinstance(membership_person_dict_obj, dict):
membership_person_dict = membership_person_dict_obj
try:
membership_person_obj = Membership_Person_Base(**membership_person_dict_obj)
membership_person_obj = Membership_Person_Base(**membership_person_dict)
log.debug(membership_person_obj)
except ValidationError as e:
log.error(e.json())
return False
else:
membership_person_obj = membership_person_dict_obj # Assume it is an object model
membership_person_obj = membership_person_dict_obj
membership_person_obj.account_id = account_id
membership_person_obj.person_id = person_id
membership_person_dict = membership_person_dict_obj.dict(by_alias=False, exclude_defaults=False, exclude_unset=True, exclude={'membership_cfg', 'membership_group', 'membership_group_list', 'membership_person_group_list', 'profile', 'membership_person_profile', 'membership_type', 'membership_type_list', 'membership_person_type', 'person', 'product', 'user', 'created_on', 'updated_on'})
log.debug(membership_person_dict)
membership_person_dict = membership_person_obj.dict(by_alias=False, exclude_defaults=False, exclude_unset=True, exclude={'membership_cfg', 'membership_person_group_list', 'membership_person_profile', 'membership_person_type', 'membership_group', 'membership_group_list', 'membership_type', 'membership_type_list', 'person', 'product', 'created_on', 'updated_on'})
# ### SECTION ### Process data
membership_person_obj.account_id = account_id # Is this needed?
membership_person_dict['account_id'] = account_id
if membership_person_dict_in_result := sql_insert(data=membership_person_dict, table_name='membership_person', rm_id_random=True, id_random_length=8): pass
if membership_person_dict_in_result := sql_insert(
data = membership_person_dict,
table_name = 'membership_person',
rm_id_random = True,
id_random_length = default_num_bytes
): pass
else:
log.warning(f'Membership Person not created.')
log.debug(membership_person_dict_in_result)
return False
log.debug(membership_person_dict_in_result)
membership_person_id = membership_person_dict_in_result
membership_person_outline = {}
@@ -434,11 +447,45 @@ def create_membership_person_obj(
membership_person_outline['account_id'] = account_id
membership_person_outline['person_id'] = person_id
membership_person_outline['membership_group_list'] = []
membership_person_outline['membership_type'] = {}
membership_person_outline['membership_type_id'] = None
membership_person_outline['membership_person_group_list'] = []
membership_person_outline['membership_person_profile'] = {}
membership_person_outline['membership_person_profile_id'] = None
membership_person_outline['membership_person_type'] = {}
membership_person_outline['membership_person_type_id'] = None
if membership_person_obj.membership_person_type_list:
log.info('Looping through Membership Person Group object list')
for membership_person_type_obj in membership_person_type_obj_list_result:
if membership_person_type_obj.id:
log.info('Updating Membership Person Group object')
if membership_person_type_update_result := update_membership_person_type_obj(
membership_person_type_id = membership_person_type_id,
membership_person_type_dict_obj = membership_person_type_obj,
): pass
else: return False
else:
log.info('Creating Membership Person Group object')
if membership_person_type_create_result := create_membership_person_type_obj(
membership_person_id = membership_person_id,
membership_person_type_dict_obj = membership_person_type_obj,
): pass
else: pass
if membership_person_obj.membership_person_type and membership_person_obj.membership_person_type.id:
log.info('Updating Membership Person Type object')
if membership_person_type_update_result := update_membership_person_type_obj(
membership_person_type_id = membership_person_type_id,
membership_person_type_dict_obj = membership_person_obj.membership_person_type,
): pass
else: return False
elif membership_person_obj.membership_person_type:
log.info('Creating Membership Type object')
if membership_person_type_create_result := create_membership_person_type_obj(
membership_person_id = membership_person_id,
membership_person_type_dict_obj = membership_person_obj.membership_person_type,
): pass
else: return False
else: pass
if return_outline:
log.debug(f'Returning the Membership Person Outline: {membership_person_outline}')
@@ -452,37 +499,45 @@ def create_membership_person_obj(
# ### BEGIN ### API Membership Person Methods ### update_membership_person_obj() ###
# Updated 2022-01-04
def update_membership_person_obj(
membership_person_id: int|str,
membership_person_dict_obj: Membership_Person_Base|dict,
membership_person_id: int,
membership_person_dict_obj: Membership_Person_Base,
create_sub_obj: bool = False,
fail_any: bool = False, # Fail if any thing goes wrong for sub objects
return_outline: bool = False,
) -> int|bool:
) -> bool|dict:
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
# ### SECTION ### Secondary data validation
log.info('Checking requirements...')
if membership_person_id := redis_lookup_id_random(record_id_random=membership_person_id, table_name='membership_person'): pass
else:
log.error('Membership Person ID passed but is invalid. Failed requirement.')
return False
# ### SECTION ### Secondary data validation
log.info('Create dictionary or Pydantic object')
log.debug(type(membership_person_dict_obj))
if isinstance(membership_person_dict_obj, dict):
membership_person_dict = membership_person_dict_obj
try:
membership_person_obj = Membership_Person_Base(**membership_person_dict_obj)
# membership_person_obj.id = membership_person_id
membership_person_obj = Person_Base(**membership_person_dict)
log.debug(membership_person_obj)
except ValidationError as e:
log.error(e.json())
return False
else:
membership_person_obj = membership_person_dict_obj # Assume it is an object model
membership_person_obj = membership_person_dict_obj
membership_person_dict = membership_person_dict_obj.dict(by_alias=False, exclude_defaults=False, exclude_unset=True, exclude={'membership_cfg', 'membership_group', 'membership_group_list', 'membership_person_group_list', 'profile', 'membership_person_profile', 'membership_type', 'membership_type_list', 'membership_person_type', 'person', 'product', 'user', 'created_on', 'updated_on'})
log.debug(membership_person_dict)
membership_person_dict = membership_person_obj.dict(by_alias=False, exclude_defaults=False, exclude_unset=True, exclude={'membership_cfg', 'membership_person_group_list', 'membership_person_profile', 'membership_person_type', 'membership_group', 'membership_group_list', 'membership_type', 'membership_type_list', 'person', 'product', 'created_on', 'updated_on'})
if membership_person_dict_up_result := sql_update(data=membership_person_dict, table_name='membership_person', rm_id_random=True, id_random_length=8): pass
# ### SECTION ### Process data
if membership_person_dict_up_result := sql_update(
data = membership_person_dict,
table_name = 'membership_person',
rm_id_random = True,
): pass
else:
log.warning(f'Membership Person not updated.')
log.debug(membership_person_dict_up_result)
@@ -490,56 +545,90 @@ def update_membership_person_obj(
membership_person_outline = {}
membership_person_outline['membership_person_id'] = membership_person_id
membership_person_outline['account_id'] = membership_person_obj.account_id_random
membership_person_outline['person_id'] = membership_person_obj.person_id_random
membership_person_outline['account_id'] = account_id
membership_person_outline['person_id'] = person_id
membership_person_outline['membership_group_list'] = []
membership_person_outline['membership_type'] = {}
membership_person_outline['membership_type_id'] = None
membership_person_outline['membership_person_group_list'] = []
membership_person_outline['membership_person_profile'] = {}
membership_person_outline['membership_person_profile_id'] = None
membership_person_outline['membership_person_type'] = {}
membership_person_outline['membership_person_type_id'] = None
# NOTE: Use object model version because of better type checking and validations
if membership_person_obj.membership_person_type:
membership_person_outline['membership_person_type_id'] = None
membership_person_type_obj = membership_person_obj.membership_person_type
if membership_person_type_id := membership_person_obj.membership_person_type_id: pass
elif membership_person_type_id := membership_person_type_obj.id: pass
else: membership_person_type_id = None
membership_person_type_obj.membership_type_id = 6
if membership_person_type_id:
update_membership_person_type_obj_result = update_membership_person_type_obj(
membership_person_type_dict_obj = membership_person_type_obj,
membership_person_type_id = membership_person_type_id,
fail_any = fail_any,
return_outline = return_outline,
)
if update_membership_person_type_obj_result:
membership_person_outline['membership_person_type_id'] = membership_person_type_id
pass
if membership_person_obj.membership_person_type_list:
log.info('Looping through Membership Person Group object list')
for membership_person_type_obj in membership_person_type_obj_list_result:
if membership_person_type_obj.id:
log.info('Updating Membership Person Group object')
if membership_person_type_update_result := update_membership_person_type_obj(
membership_person_type_id = membership_person_type_id,
membership_person_type_dict_obj = membership_person_type_obj,
): pass
else: return False
else:
pass
else:
create_membership_person_type_obj_result = create_membership_person_type_obj(
membership_person_type_dict_obj = membership_person_type_obj,
log.info('Creating Membership Person Group object')
if membership_person_type_create_result := create_membership_person_type_obj(
membership_person_id = membership_person_id,
membership_person_type_dict_obj = membership_person_type_obj,
): pass
else: pass
if membership_person_obj.membership_person_type and membership_person_obj.membership_person_type.id:
log.info('Updating Membership Person Type object')
if membership_person_type_update_result := update_membership_person_type_obj(
membership_person_type_id = membership_person_type_id,
membership_person_type_dict_obj = membership_person_obj.membership_person_type,
): pass
else: return False
elif membership_person_obj.membership_person_type:
log.info('Creating Membership Type object')
if membership_person_type_create_result := create_membership_person_type_obj(
membership_person_id = membership_person_id,
fail_any = fail_any,
return_outline = return_outline,
)
if isinstance(create_membership_person_type_obj_result, int):
membership_person_type_id = create_membership_person_type_obj_result
membership_person_outline['membership_person_type_id'] = membership_person_type_id
else:
pass
membership_person_type_dict_obj = membership_person_obj.membership_person_type,
): pass
else: return False
else: pass
# # NOTE: Use object model version because of better type checking and validations
# if membership_person_obj.membership_person_type:
# membership_person_outline['membership_person_type_id'] = None
# membership_person_type_obj = membership_person_obj.membership_person_type
# if membership_person_type_id := membership_person_obj.membership_person_type_id: pass
# elif membership_person_type_id := membership_person_type_obj.id: pass
# else: membership_person_type_id = None
# membership_person_type_obj.membership_type_id = 6
# if membership_person_type_id:
# update_membership_person_type_obj_result = update_membership_person_type_obj(
# membership_person_type_dict_obj = membership_person_type_obj,
# membership_person_type_id = membership_person_type_id,
# fail_any = fail_any,
# return_outline = return_outline,
# )
# if update_membership_person_type_obj_result:
# membership_person_outline['membership_person_type_id'] = membership_person_type_id
# pass
# else:
# pass
# else:
# create_membership_person_type_obj_result = create_membership_person_type_obj(
# membership_person_type_dict_obj = membership_person_type_obj,
# membership_person_id = membership_person_id,
# fail_any = fail_any,
# return_outline = return_outline,
# )
# if isinstance(create_membership_person_type_obj_result, int):
# membership_person_type_id = create_membership_person_type_obj_result
# membership_person_outline['membership_person_type_id'] = membership_person_type_id
# else:
# pass
if return_outline:
log.debug(f'Returning the Membership Person Outline: {membership_person_outline}')
return membership_person_outline
else:
log.debug(f'Returning the Membership Person ID: {membership_person_id}')
return membership_person_id
log.debug(f'Returning True')
return True
# ### END ### API Membership Person Methods ### create_update_membership_person_obj() ###

78
app/methods/membership_person_type_methods.py
View File

@@ -195,6 +195,7 @@ def create_membership_person_type_obj(
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
# ### SECTION ### Secondary data validation
log.info('Checking requirements...')
if membership_person_id := redis_lookup_id_random(record_id_random=membership_person_id, table_name='membership_person'): pass
else:
@@ -202,27 +203,42 @@ def create_membership_person_type_obj(
log.info(f'Membership Person ID: {membership_person_id}')
return False
membership_person_type_obj = membership_person_type_dict_obj
log.info('Create dictionary or Pydantic object')
log.debug(type(membership_person_type_dict_obj))
if isinstance(membership_person_type_dict_obj, dict):
membership_person_type_dict = membership_person_type_dict_obj
try:
membership_person_type_obj = Membership_Person_Type_Base(**membership_person_type_dict)
log.debug(membership_person_type_obj)
except ValidationError as e:
log.error(e.json())
return False
else:
membership_person_type_obj = membership_person_type_dict_obj
membership_person_type_obj.membership_person_id = membership_person_id
# Stuff goes here
membership_person_type_dict = membership_person_type_obj.dict(by_alias=False, exclude_defaults=False, exclude_unset=True, exclude={'membership_cfg', 'membership_type', 'membership_type_list', 'product', 'created_on', 'updated_on'})
membership_person_type_obj_data = membership_person_type_obj.dict(by_alias=False, exclude_defaults=False, exclude_unset=True, exclude={'created_on', 'updated_on'})
# ### SECTION ### Process data
membership_person_type_obj.membership_person_id = membership_person_id # Is this needed?
membership_person_type_dict['membership_person_id'] = membership_person_id
membership_person_type_obj_data['membership_person_id'] = membership_person_id
if membership_person_type_obj_in_result := sql_insert(
data = membership_person_type_obj_data,
if membership_person_type_dict_in_result := sql_insert(
data = membership_person_type_dict,
table_name = 'membership_person_type',
rm_id_random = True,
id_random_length = 8
id_random_length = default_num_bytes
): pass
else: return False
else:
log.warning(f'Membership Person Type not created.')
log.debug(membership_person_type_dict_in_result)
return False
log.debug(membership_person_type_obj_in_result)
log.debug(membership_person_type_dict_in_result)
membership_person_type_id = membership_person_type_dict_in_result
membership_person_type_id = membership_person_type_obj_in_result
log.info(f'Returning the Membership Person Type ID: {membership_person_type_id}')
log.debug(f'New membership_person_type_id: {membership_person_type_id}')
return membership_person_type_id
# ### END ### API Membership Type Person Methods ### create_membership_person_type_obj() ###
@@ -237,15 +253,31 @@ def update_membership_person_type_obj(
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
# ### SECTION ### Secondary data validation
log.info('Checking requirements...')
if membership_person_type_id := redis_lookup_id_random(record_id_random=membership_person_type_id, table_name='membership_person_type'): pass
else: return False
else:
log.error('Membership Person Type ID passed but is invalid. Failed requirement.')
return False
# Stuff goes here
log.info('Create dictionary or Pydantic object')
log.debug(type(membership_person_type_dict_obj))
if isinstance(membership_person_type_dict_obj, dict):
membership_person_type_dict = membership_person_type_dict_obj
try:
membership_person_type_obj = Membership_Person_Type_Base(**membership_person_type_dict)
log.debug(membership_person_type_obj)
except ValidationError as e:
log.error(e.json())
return False
else:
membership_person_type_obj = membership_person_type_dict_obj
membership_person_type_dict_obj.id = membership_person_type_id
membership_person_type_dict = membership_person_type_obj.dict(by_alias=False, exclude_defaults=False, exclude_unset=True, exclude={'membership_cfg', 'membership_type', 'membership_type_list', 'product', 'created_on', 'updated_on'})
membership_person_type_obj = membership_person_type_dict_obj
# ### SECTION ### Process data
membership_person_type_obj.id = membership_person_type_id # Is this needed?
membership_person_type_dict['id'] = membership_person_type_id
log.debug(membership_person_type_obj)
log.debug(membership_person_type_obj.dict(by_alias=False, exclude_unset=True))
@@ -254,10 +286,16 @@ def update_membership_person_type_obj(
membership_person_type_dict = membership_person_type_obj.dict(by_alias=False, exclude_unset=True)
log.debug(membership_person_type_dict)
if membership_person_type_obj_up_result := sql_update(data=membership_person_type_dict, table_name='membership_person_type', rm_id_random=True):
log.debug(membership_person_type_obj_up_result)
if membership_person_type_dict_up_result := sql_update(
data = membership_person_type_dict,
table_name = 'membership_person_type',
rm_id_random = True
):
log.debug(f'Returning True')
log.debug(membership_person_type_dict_up_result)
return True
else:
log.debug(membership_person_type_obj_up_result)
log.warning(f'Membership Person Type not updated.')
log.debug(membership_person_type_dict_up_result)
return False
# ### END ### API Membership Type Person Methods ### update_membership_person_type_obj() ###

28
app/methods/person_methods.py
View File

@@ -452,7 +452,7 @@ def create_person_kiss(
contact_id: int|None = None,
organization_id: int|None = None,
user_id: int|None = None,
):
) -> int|bool:
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
@@ -469,13 +469,14 @@ def create_person_kiss(
return False
else:
person_obj = person_dict_obj
person_obj.account_id = account_id
# person_obj.account_id = account_id
person_dict = person_obj.dict(by_alias=False, exclude_defaults=False, exclude_unset=True, exclude={'contact', 'contact_id', 'contact_id_random', 'email', 'cc_email', 'membership_person_id', 'membership_person_id_random', 'organization', 'user', 'created_on', 'updated_on'})
# ### SECTION ### Process data# Look for an account_id in the user_obj
if account_id: pass
elif account_id := person_obj.account_id: pass
# ### SECTION ### Process data
# Look for an account_id in the person_obj
# if account_id: pass
# elif account_id := person_obj.account_id: pass
person_obj.account_id = account_id # Is this needed?
person_dict['account_id'] = account_id
@@ -493,7 +494,12 @@ def create_person_kiss(
# person_obj.user_id = user_id # Is this needed?
# person_dict['user_id'] = user_id
if person_dict_in_result := sql_insert(data=person_dict, table_name='person', rm_id_random=True, id_random_length=default_num_bytes): pass
if person_dict_in_result := sql_insert(
data = person_dict,
table_name = 'person',
rm_id_random = True,
id_random_length = default_num_bytes
): pass
else:
log.warning(f'Person not created.')
log.debug(person_dict_in_result)
@@ -548,7 +554,7 @@ def create_person_kiss(
else: return False
else: pass
log.info(f'Returning the new person_id: {person_id}')
log.info(f'Returning the Person ID: {person_id}')
return person_id
# ### END ### API Person Methods ### create_person_kiss() ###
@@ -573,7 +579,6 @@ def update_person_kiss(
person_dict = person_dict_obj
try:
person_obj = Person_Base(**person_dict)
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(person_obj)
except ValidationError as e:
log.error(e.json())
@@ -600,7 +605,12 @@ def update_person_kiss(
# person_obj.user_id = user_id # Is this needed?
# person_dict['user_id'] = user_id
if person_dict_up_result := sql_update(data=person_dict, table_name='person', rm_id_random=True, id_random_length=default_num_bytes): pass
if person_dict_up_result := sql_update(
data = person_dict,
table_name = 'person',
rm_id_random = True,
id_random_length = default_num_bytes
): pass
else:
log.warning(f'Person not updated.')
log.debug(person_dict_up_result)