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

Code clean up and standardize

Browse Source
This commit is contained in:
Scott Idem
2021-08-25 10:58:39 -04:00
parent 7859d1d2b2
commit 1369874dc2
32 changed files with 479 additions and 195 deletions
Download Patch File Download Diff File Expand all files Collapse all files

260
app/methods/person_methods.py
View File

@@ -339,6 +339,37 @@ def get_person_rec_w_external_id(
# ### END ### API Person Methods ### get_person_rec_w_external_id() ###
# ### BEGIN ### API Person Methods ### get_account_id_w_person_id() ###
# Updated 2021-08-25
def get_account_id_w_person_id(
person_id: int|str,
) -> bool|int|None:
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
if person_id := redis_lookup_id_random(record_id_random=person_id, table_name='person'): pass
else: return False
data = {}
data['person_id'] = person_id
sql = f"""
SELECT `person`.id AS 'person_id', `person`.id_random AS 'person_id_random', `person`.account_id AS account_id
FROM `person` AS `person`
WHERE `person`.id = :person_id
LIMIT 1;
"""
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
if person_data_result := sql_select(data=data, sql=sql):
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(person_data_result)
if account_id := person_data_result.get('account_id', None): return account_id
else: return False
else: return None
# ### END ### API Person Methods ### get_account_id_w_person_id() ###
# ### BEGIN ### API Person Methods ### create_person_obj_v3() ###
# NOTE: This will create a person and then also create a linked contact if person_obj.contact data is passed. The create_contact_obj will create a contact and then also create a linked address if person_obj.contact.address data is passed.
# Reviewed and updated 2021-08-24
@@ -373,6 +404,7 @@ def create_person_obj_v3(
return_dict['contact_id'] = None
return_dict['contact'] = {}
return_dict['contact']['address_id'] = None
return_dict['organization_id'] = None
return_dict['user_id'] = None
if person_obj_new.contact and isinstance(person_obj_new.contact, dict):
@@ -386,7 +418,7 @@ def create_person_obj_v3(
if update_contact_obj_result := update_contact_obj(
contact_id = contact_id,
contact_obj_up = contact_obj_unknown,
create_missing_obj = create_sub_obj,
create_sub_obj = create_sub_obj,
fail_any = fail_any,
):
contact_id = update_contact_obj_result
@@ -463,7 +495,7 @@ def create_person_obj_v3(
if update_organization_obj_result := update_organization_obj(
organization_id = organization_id,
organization_obj_up = organization_obj_unknown,
create_missing_obj = create_sub_obj,
create_sub_obj = create_sub_obj,
fail_any = fail_any,
):
organization_id = update_organization_obj_result
@@ -508,9 +540,6 @@ def create_person_obj_v3(
log.info('Organization not found or not in a dict.')
pass
if person_obj_new.user and isinstance(person_obj_new.user, dict):
log.info(f'User was found. Create a new User and link it to the new Person or update existing User. Person ID: {person_id}')
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
@@ -522,7 +551,7 @@ def create_person_obj_v3(
if update_user_obj_result := update_user_obj_v3(
user_id = user_id,
user_obj_up = user_obj_unknown,
# create_missing_obj = create_sub_obj,
# create_sub_obj = create_sub_obj,
# fail_any = fail_any,
):
user_id = update_user_obj_result
@@ -616,13 +645,220 @@ def create_person_obj_v3(
# ### END ### API Person Methods ### create_person_obj_v3() ###
# ### BEGIN ### API Person Methods ### update_person_obj_v3() ###
# Updated 2021-08-24
def update_person_obj_v3(
person_id: int|str,
person_obj_exist: Event_Person_Base,
create_sub_obj: bool = False,
fail_any: bool = False, # Fail if any thing goes wrong for sub objects
) -> bool:
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
if person_id := redis_lookup_id_random(record_id_random=person_id, table_name='person'): pass
else: return False
# Can't update the person_id alias if the .id was never set.
# person_obj_exist.person_id = person_id
if not person_obj_exist.id:
person_obj_exist.id = person_id
person_obj_data = person_obj_exist.dict(by_alias=False, exclude_defaults=False, exclude_unset=True, exclude={'contact', 'organization', 'user', 'created_on', 'updated_on'})
log.debug(person_obj_data)
if person_obj_up_result := sql_update(data=person_obj_data, table_name='person', rm_id_random=True): pass
else:
log.warning(f'Person not updated.')
log.debug(person_obj_up_result)
return False
return_dict = {}
return_dict['person_id'] = person_id
return_dict['contact_id'] = None
return_dict['contact'] = {}
return_dict['contact']['address_id'] = None
return_dict['organization_id'] = None
return_dict['user_id'] = None
if person_obj_exist.contact and isinstance(person_obj_exist.contact, dict):
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
contact_obj_unknown = person_obj_exist.contact
log.debug(contact_obj_unknown)
if contact_id := contact_obj_unknown.get('contact_id_random', None):
from app.methods.contact_methods import update_contact_obj_v3
if update_contact_obj_result := update_contact_obj_v3(
contact_id = contact_id,
contact_obj_exist = contact_obj_unknown,
create_sub_obj = create_sub_obj,
fail_any = fail_any,
):
contact_id = update_contact_obj_result
log.info(f'Contact updated. Contact ID: {contact_id}')
else:
log.warning(f'Contact not updated. Person ID: {person_id}')
log.debug(update_contact_obj_result)
contact_id = None
if fail_any: return False
if isinstance(update_contact_obj_result, int):
contact_id = update_contact_obj_result
log.info(f'Contact updated. Contact ID: {contact_id}')
else:
log.warning(f'Contact not updated. Person ID: {person_id}')
log.debug(update_contact_obj_result)
contact_id = None
if fail_any: return False
else:
log.info(f'No Contact ID found.')
from app.methods.contact_methods import create_contact_obj
if create_contact_obj_result := create_contact_obj(
person_id = person_id,
contact_obj_new = contact_obj_unknown,
create_sub_obj = create_sub_obj,
fail_any = fail_any,
):
if isinstance(create_contact_obj_result, int):
contact_id = create_contact_obj_result
log.info(f'Contact created. Contact ID: {contact_id}')
else:
log.warning(f'Contact not created. Person ID: {person_id}')
log.debug(create_contact_obj_result)
contact_id = None
if fail_any: return False
else:
log.warning(f'Contact not created. Person ID: {person_id}')
log.debug(create_contact_obj_result)
contact_id = None
if fail_any: return False
return_dict['contact_id'] = contact_id
else:
log.info('Contact not found or not in a dict.')
pass
if person_obj_exist.organization and isinstance(person_obj_exist.organization, dict):
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
organization_obj_unknown = person_obj_exist.organization
log.debug(organization_obj_unknown)
if organization_id := organization_obj_unknown.get('organization_id_random', None):
# from app.methods.organization_methods import update_organization_obj_v3
if update_organization_obj_result := update_organization_obj(
organization_id = organization_id,
organization_obj_exist = organization_obj_unknown,
create_sub_obj = create_sub_obj,
fail_any = fail_any,
):
organization_id = update_organization_obj_result
log.info(f'Organization updated. Organization ID: {organization_id}')
else:
log.warning(f'Organization not updated. Person ID: {person_id}')
log.debug(update_organization_obj_result)
organization_id = None
if fail_any: return False
if isinstance(update_organization_obj_result, int):
organization_id = update_organization_obj_result
log.info(f'Organization updated. Organization ID: {organization_id}')
else:
log.warning(f'Organization not updated. Person ID: {person_id}')
log.debug(update_organization_obj_result)
organization_id = None
if fail_any: return False
else:
log.info(f'No Organization ID found.')
# from app.methods.organization_methods import create_organization_obj_v3
if create_organization_obj_result := create_organization_obj(
person_id = person_id,
organization_obj_new = organization_obj_unknown,
create_sub_obj = create_sub_obj,
fail_any = fail_any,
):
if isinstance(create_organization_obj_result, int):
organization_id = create_organization_obj_result
log.info(f'Organization created. Organization ID: {organization_id}')
else:
log.warning(f'Organization not created. Person ID: {person_id}')
log.debug(create_organization_obj_result)
organization_id = None
if fail_any: return False
else:
log.warning(f'Organization not created. Person ID: {person_id}')
log.debug(create_organization_obj_result)
organization_id = None
if fail_any: return False
return_dict['organization_id'] = organization_id
else:
log.info('Organization not found or not in a dict.')
pass
if person_obj_exist.user and isinstance(person_obj_exist.user, dict):
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
user_obj_unknown = person_obj_exist.user
log.debug(user_obj_unknown)
if user_id := user_obj_unknown.get('user_id_random', None):
# from app.methods.user_methods import update_user_obj_v3
if update_user_obj_result := update_user_obj(
user_id = user_id,
user_obj_exist = user_obj_unknown,
create_sub_obj = create_sub_obj,
fail_any = fail_any,
):
user_id = update_user_obj_result
log.info(f'User updated. User ID: {user_id}')
else:
log.warning(f'User not updated. Person ID: {person_id}')
log.debug(update_user_obj_result)
user_id = None
if fail_any: return False
if isinstance(update_user_obj_result, int):
user_id = update_user_obj_result
log.info(f'User updated. User ID: {user_id}')
else:
log.warning(f'User not updated. Person ID: {person_id}')
log.debug(update_user_obj_result)
user_id = None
if fail_any: return False
else:
log.info(f'No User ID found.')
# from app.methods.user_methods import create_user_obj_v3
if create_user_obj_result := create_user_obj(
person_id = person_id,
user_obj_new = user_obj_unknown,
create_sub_obj = create_sub_obj,
fail_any = fail_any,
):
if isinstance(create_user_obj_result, int):
user_id = create_user_obj_result
log.info(f'User created. User ID: {user_id}')
else:
log.warning(f'User not created. Person ID: {person_id}')
log.debug(create_user_obj_result)
user_id = None
if fail_any: return False
else:
log.warning(f'User not created. Person ID: {person_id}')
log.debug(create_user_obj_result)
user_id = None
if fail_any: return False
return_dict['user_id'] = user_id
else:
log.info('User not found or not in a dict.')
pass
log.info(f'The Person has been updated. Person ID: {person_id}')
return True
# ### END ### API Person Methods ### update_person_obj_v3() ###
# ### BEGIN ### API Person Methods ### update_person_obj() ###
# NOTE: This will update a person and then also create or update a linked contact if person_obj.contact data is passed. The create_contact_obj will create a contact and then also create a linked address if person_obj.contact.address data is passed. The update_contact_obj will update a contact and then also create or update a linked address if person_obj.contact.address data is passed.
# Reviewed and updated 2021-08-10
def update_person_obj(
person_id: int|str, # Ideally the int ID should be passed. This allows for updating of the id_random value.
person_obj_up: Person_Base,
create_missing_obj: bool = False,
create_sub_obj: bool = False,
fail_any: bool = True, # Fail if any thing goes wrong for sub objects
) -> bool:
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
@@ -646,7 +882,7 @@ def update_person_obj(
if contact_obj_up_result := update_contact_obj(
contact_id = contact_id,
contact_obj_up = contact_obj_up,
create_missing_obj = create_missing_obj,
create_sub_obj = create_sub_obj,
):
log.debug(contact_obj_up_result)
else:
@@ -675,7 +911,7 @@ def update_person_obj(
if organization_obj_up_result := update_organization_obj(
organization_id = organization_id,
organization_obj_up = organization_obj_up,
create_missing_obj = create_missing_obj,
create_sub_obj = create_sub_obj,
):
log.debug(organization_obj_up_result)
else:
@@ -702,9 +938,9 @@ def update_person_obj(
log.debug(user_id)
log.debug(user_obj_up)
if user_obj_up_result := update_user_obj(
user_id=user_id,
user_obj_up=user_obj_up,
create_missing_obj=create_missing_obj,
user_id = user_id,
user_obj_up = user_obj_up,
create_sub_obj = create_sub_obj,
):
log.debug(user_obj_up_result)
else: