Code clean up. Simplifying person, user, contact, and address methods
This commit is contained in:
Scott Idem
@@ -313,10 +313,10 @@ def create_update_contact_obj_v4(
|
||||
# ### BEGIN ### API Contact Methods ### create_contact_obj() ###
|
||||
# NOTE: This will create a contact and then also create a linked address if contact_obj.address data is passed.
|
||||
# NOTE: In the future it should be required that account_id, for_type, and for_id should be passed separately. account_id might not be required *if* it can be looked up based on for_type and for_id.
|
||||
# Reviewed and updated 2021-08-24
|
||||
# Updated 2022-01-06
|
||||
def create_contact_obj(
|
||||
contact_obj_new: Contact_Base,
|
||||
account_id: int|str = None,
|
||||
account_id: int|str,
|
||||
contact_dict_obj: Contact_Base,
|
||||
for_type: str = None,
|
||||
for_id: int|str = None,
|
||||
create_sub_obj: bool = False,
|
||||
@@ -325,82 +325,128 @@ def create_contact_obj(
|
||||
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
|
||||
log.debug(locals())
|
||||
|
||||
# ### SECTION ### Secondary data validation
|
||||
# NOTE: Remove at future date. Is this check needed if we trust that the ID is checked ahead of time?
|
||||
if account_id := redis_lookup_id_random(record_id_random=account_id, table_name='account'): pass
|
||||
else: return False
|
||||
|
||||
if for_id := redis_lookup_id_random(record_id_random=for_id, table_name=for_type): pass
|
||||
else: return False
|
||||
|
||||
log.debug(type(contact_obj_new))
|
||||
if isinstance(contact_obj_new, dict):
|
||||
if account_id:
|
||||
contact_obj_new['account_id'] = account_id
|
||||
if for_type:
|
||||
contact_obj_new['for_type'] = for_type
|
||||
if for_id:
|
||||
contact_obj_new['for_id'] = for_id
|
||||
log.info('Create dictionary or Pydantic object')
|
||||
log.debug(type(contact_dict_obj))
|
||||
if isinstance(contact_dict_obj, dict):
|
||||
contact_dict = contact_dict_obj
|
||||
try:
|
||||
contact_obj_new = Contact_Base(**contact_obj_new)
|
||||
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
|
||||
log.debug(contact_obj_new)
|
||||
contact_obj = Contact_Base(**contact_dict)
|
||||
log.debug(contact_obj)
|
||||
except ValidationError as e:
|
||||
log.error(e.json())
|
||||
return False
|
||||
else:
|
||||
if account_id:
|
||||
contact_obj_new.account_id = account_id
|
||||
if for_type:
|
||||
contact_obj_new.for_type = for_type
|
||||
if for_id:
|
||||
contact_obj_new.for_id = for_id
|
||||
contact_obj = contact_dict_obj
|
||||
contact_obj.account_id = account_id
|
||||
|
||||
contact_obj_data = contact_obj_new.dict(by_alias=False, exclude_defaults=False, exclude_unset=True, exclude={'address', 'created_on', 'updated_on'})
|
||||
contact_dict = contact_obj.dict(by_alias=False, exclude_defaults=False, exclude_unset=True, exclude={'address', 'address_id', 'address_id_random', 'created_on', 'updated_on'})
|
||||
|
||||
if contact_obj_in_result := sql_insert(data=contact_obj_data, table_name='contact', rm_id_random=True, id_random_length=8): pass
|
||||
# log.debug(type(contact_dict_obj))
|
||||
# if isinstance(contact_dict_obj, dict):
|
||||
# if account_id:
|
||||
# contact_dict_obj['account_id'] = account_id
|
||||
# if for_type:
|
||||
# contact_dict_obj['for_type'] = for_type
|
||||
# if for_id:
|
||||
# contact_dict_obj['for_id'] = for_id
|
||||
# try:
|
||||
# contact_obj = Contact_Base(**contact_dict_obj)
|
||||
# log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
|
||||
# log.debug(contact_obj)
|
||||
# except ValidationError as e:
|
||||
# log.error(e.json())
|
||||
# return False
|
||||
# else:
|
||||
# if account_id:
|
||||
# contact_obj.account_id = account_id
|
||||
# if for_type:
|
||||
# contact_obj.for_type = for_type
|
||||
# if for_id:
|
||||
# contact_obj.for_id = for_id
|
||||
|
||||
# contact_dict = contact_obj.dict(by_alias=False, exclude_defaults=False, exclude_unset=True, exclude={'address', 'created_on', 'updated_on'})
|
||||
|
||||
# ### SECTION ### Process data
|
||||
contact_obj.account_id = account_id
|
||||
contact_obj.for_type = for_type
|
||||
contact_obj.for_id = for_id
|
||||
contact_dict['account_id'] = account_id
|
||||
contact_dict['for_type'] = for_type
|
||||
contact_dict['for_id'] = for_id
|
||||
|
||||
if contact_dict_in_result := sql_insert(data=contact_dict, table_name='contact', rm_id_random=True, id_random_length=8): pass
|
||||
else:
|
||||
return False
|
||||
|
||||
log.debug(contact_obj_in_result)
|
||||
log.debug(contact_dict_in_result)
|
||||
contact_id = contact_dict_in_result
|
||||
|
||||
contact_id = contact_obj_in_result
|
||||
|
||||
if contact_obj_new.address:
|
||||
address_obj_new = contact_obj_new.address
|
||||
address_obj_new.for_type = 'contact'
|
||||
address_obj_new.for_id = contact_id
|
||||
create_address_obj_result = create_address_obj(
|
||||
address_obj_new = address_obj_new,
|
||||
account_id = account_id,
|
||||
if address_id and contact.address:
|
||||
log.info('Updating Address object')
|
||||
if address_update_result := update_address_obj(
|
||||
address_id = address_id,
|
||||
address_dict_obj = contact_obj.address,
|
||||
for_type = 'contact',
|
||||
for_id = contact_id,
|
||||
fail_any = fail_any,
|
||||
)
|
||||
if isinstance(create_address_obj_result, int):
|
||||
address_id = create_address_obj_result
|
||||
# NOTE: This last update should no longer be needed now that the contact.address_id is not supposed to be used.
|
||||
# Need to update the contact with the new address_id
|
||||
contact_obj_up = {} # REMOVE
|
||||
contact_obj_up['id'] = contact_id # REMOVE
|
||||
contact_obj_up['address_id'] = address_id # REMOVE
|
||||
if contact_obj_up_result := sql_update(data=contact_obj_up, table_name='contact'): pass # REMOVE
|
||||
else: # REMOVE
|
||||
return False # REMOVE
|
||||
log.debug(contact_obj_up_result) # REMOVE
|
||||
else:
|
||||
log.debug(f'No address_id was returned when tyring to create_address_obj(): {create_address_obj_result}')
|
||||
address_id = None
|
||||
): pass
|
||||
else: return False
|
||||
elif contact.address:
|
||||
log.info('Creating Address object')
|
||||
if address_create_result := create_address_obj(
|
||||
account_id = account_id,
|
||||
address_dict_obj = contact_obj.address,
|
||||
for_type = 'contact',
|
||||
for_id = contact_id,
|
||||
): pass
|
||||
else: return False
|
||||
else: pass
|
||||
|
||||
# if contact_obj.address:
|
||||
# address_obj_new = contact_obj.address
|
||||
# address_obj_new.for_type = 'contact'
|
||||
# address_obj_new.for_id = contact_id
|
||||
# create_address_obj_result = create_address_obj(
|
||||
# address_obj_new = address_obj_new,
|
||||
# account_id = account_id,
|
||||
# for_type = 'contact',
|
||||
# for_id = contact_id,
|
||||
# fail_any = fail_any,
|
||||
# )
|
||||
# if isinstance(create_address_obj_result, int):
|
||||
# address_id = create_address_obj_result
|
||||
# # NOTE: This last update should no longer be needed now that the contact.address_id is not supposed to be used.
|
||||
# # Need to update the contact with the new address_id
|
||||
# contact_obj_up = {} # REMOVE
|
||||
# contact_obj_up['id'] = contact_id # REMOVE
|
||||
# contact_obj_up['address_id'] = address_id # REMOVE
|
||||
# if contact_obj_up_result := sql_update(data=contact_obj_up, table_name='contact'): pass # REMOVE
|
||||
# else: # REMOVE
|
||||
# return False # REMOVE
|
||||
# log.debug(contact_obj_up_result) # REMOVE
|
||||
# else:
|
||||
# log.debug(f'No address_id was returned when tyring to create_address_obj(): {create_address_obj_result}')
|
||||
# address_id = None
|
||||
|
||||
log.info(f'Returning the new contact_id: {contact_id}')
|
||||
|
||||
log.debug(f'Returning the new contact_id: {contact_id}')
|
||||
return contact_id
|
||||
# ### END ### API Contact Methods ### create_contact_obj() ###
|
||||
|
||||
|
||||
# ### BEGIN ### API Contact Methods ### update_contact_obj() ###
|
||||
# NOTE: This will update a contact and then also create or update a linked address if contact_obj.address data is passed.
|
||||
# Reviewed and updated 2021-08-10
|
||||
# Updated 2022-01-06
|
||||
def update_contact_obj(
|
||||
contact_id: int|str, # Ideally the int ID should be passed. This allows for updating of the id_random value.
|
||||
contact_obj_up: Contact_Base,
|
||||
contact_dict_obj: Contact_Base,
|
||||
create_sub_obj: bool = False,
|
||||
fail_any: bool = False, # Fail if any thing goes wrong for sub objects
|
||||
return_dict: bool = False,
|
||||
@@ -408,80 +454,128 @@ def update_contact_obj(
|
||||
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
|
||||
log.debug(locals())
|
||||
|
||||
# ### SECTION ### Secondary data validation
|
||||
# NOTE: Remove at future date. Is this check needed if we trust that the ID is checked ahead of time?
|
||||
if contact_id := redis_lookup_id_random(record_id_random=contact_id, table_name='contact'): pass
|
||||
else: return False
|
||||
|
||||
log.info('Create dictionary or Pydantic object')
|
||||
log.debug(type(contact_dict_obj))
|
||||
if isinstance(contact_dict_obj, dict):
|
||||
contact_dict = contact_dict_obj
|
||||
try:
|
||||
contact_obj = Contact_Base(**contact_dict)
|
||||
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
|
||||
log.debug(contact_obj)
|
||||
except ValidationError as e:
|
||||
log.error(e.json())
|
||||
return False
|
||||
else:
|
||||
contact_obj = contact_dict_obj
|
||||
|
||||
contact_dict = contact_obj.dict(by_alias=False, exclude_defaults=False, exclude_unset=True, exclude={'address', 'address_id', 'address_id_random', 'created_on', 'updated_on'})
|
||||
|
||||
# ### SECTION ### Process data
|
||||
contact_obj.id = contact_id # Is this needed?
|
||||
contact_dict['contact_id'] = contact_id
|
||||
|
||||
account_id = get_account_id_w_contact_id(contact_id)
|
||||
|
||||
contact_obj_up.id = contact_id
|
||||
if contact_dict_up_result := sql_update(data=contact_dict, table_name='contact', rm_id_random=True): pass
|
||||
else:
|
||||
log.warning(f'Contact not updated.')
|
||||
log.debug(contact_dict_up_result)
|
||||
return False
|
||||
|
||||
log.debug(contact_obj_up)
|
||||
log.debug(contact_dict_up_result)
|
||||
|
||||
if address_id and contact.address:
|
||||
log.info('Updating Address object')
|
||||
if address_update_result := update_address_obj(
|
||||
address_id = address_id,
|
||||
address_dict_obj = contact_obj.address,
|
||||
for_type = 'contact',
|
||||
for_id = contact_id,
|
||||
): pass
|
||||
else: return False
|
||||
elif contact.address:
|
||||
log.info('Creating Address object')
|
||||
if address_create_result := create_address_obj(
|
||||
account_id = account_id,
|
||||
address_dict_obj = contact_obj.address,
|
||||
for_type = 'contact',
|
||||
for_id = contact_id,
|
||||
): pass
|
||||
else: return False
|
||||
else: pass
|
||||
|
||||
# log.debug(contact_obj_up)
|
||||
# log.debug(contact_obj_up.dict(by_alias=True, exclude_unset=True))
|
||||
log.debug(contact_obj_up.dict(by_alias=False, exclude_unset=True))
|
||||
log.debug(contact_obj.dict(by_alias=False, exclude_unset=True))
|
||||
# log.debug(contact_obj_up.dict(by_alias=False, exclude_unset=False))
|
||||
|
||||
#contact_dict_up = contact_obj_up.dict(by_alias=False, exclude_unset=True)
|
||||
|
||||
if contact_obj_up.address:
|
||||
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
|
||||
address_obj_unknown = contact_obj_up.address
|
||||
log.debug(address_obj_unknown)
|
||||
if isinstance(contact_obj_up.address, dict):
|
||||
address_id = address_obj_unknown.get('address_id_random', None)
|
||||
else:
|
||||
address_id = address_obj_unknown.id
|
||||
# if contact_obj_up.address:
|
||||
# log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
|
||||
# address_obj_unknown = contact_obj_up.address
|
||||
# log.debug(address_obj_unknown)
|
||||
# if isinstance(contact_obj_up.address, dict):
|
||||
# address_id = address_obj_unknown.get('address_id_random', None)
|
||||
# else:
|
||||
# address_id = address_obj_unknown.id
|
||||
|
||||
if address_id:
|
||||
# from app.methods.address_methods import update_address_obj_v3
|
||||
if update_address_obj_result := update_address_obj(
|
||||
address_id = address_id,
|
||||
address_obj_up = address_obj_unknown,
|
||||
create_sub_obj = create_sub_obj,
|
||||
fail_any = fail_any,
|
||||
):
|
||||
address_id = update_address_obj_result
|
||||
log.info(f'Address updated. Address ID: {address_id}')
|
||||
else:
|
||||
log.warning(f'Address not updated. Contact ID: {contact_id}')
|
||||
log.debug(update_address_obj_result)
|
||||
address_id = None
|
||||
if fail_any: return False
|
||||
# if address_id:
|
||||
# # from app.methods.address_methods import update_address_obj_v3
|
||||
# if update_address_obj_result := update_address_obj(
|
||||
# address_id = address_id,
|
||||
# address_obj_up = address_obj_unknown,
|
||||
# create_sub_obj = create_sub_obj,
|
||||
# fail_any = fail_any,
|
||||
# ):
|
||||
# address_id = update_address_obj_result
|
||||
# log.info(f'Address updated. Address ID: {address_id}')
|
||||
# else:
|
||||
# log.warning(f'Address not updated. Contact ID: {contact_id}')
|
||||
# log.debug(update_address_obj_result)
|
||||
# address_id = None
|
||||
# if fail_any: return False
|
||||
|
||||
if isinstance(update_address_obj_result, int):
|
||||
address_id = update_address_obj_result
|
||||
log.info(f'Address updated. Address ID: {address_id}')
|
||||
else:
|
||||
log.warning(f'Address not updated. Contact ID: {contact_id}')
|
||||
log.debug(update_address_obj_result)
|
||||
address_id = None
|
||||
if fail_any: return False
|
||||
else:
|
||||
log.info(f'No Address ID found.')
|
||||
# from app.methods.address_methods import create_address_obj_v3
|
||||
if create_address_obj_result := create_address_obj(
|
||||
account_id = account_id,
|
||||
for_type = 'contact',
|
||||
for_id = contact_id,
|
||||
address_obj_new = address_obj_unknown,
|
||||
fail_any = fail_any,
|
||||
):
|
||||
if isinstance(create_address_obj_result, int):
|
||||
address_id = create_address_obj_result
|
||||
log.info(f'Address created. Address ID: {address_id}')
|
||||
else:
|
||||
log.warning(f'Address not created. Contact ID: {contact_id}')
|
||||
log.debug(create_address_obj_result)
|
||||
address_id = None
|
||||
if fail_any: return False
|
||||
else:
|
||||
log.warning(f'Address not created. Contact ID: {contact_id}')
|
||||
log.debug(create_address_obj_result)
|
||||
address_id = None
|
||||
if fail_any: return False
|
||||
# return_dict['address_id'] = address_id
|
||||
else:
|
||||
log.info('Address not found or not in a dict.')
|
||||
pass
|
||||
# if isinstance(update_address_obj_result, int):
|
||||
# address_id = update_address_obj_result
|
||||
# log.info(f'Address updated. Address ID: {address_id}')
|
||||
# else:
|
||||
# log.warning(f'Address not updated. Contact ID: {contact_id}')
|
||||
# log.debug(update_address_obj_result)
|
||||
# address_id = None
|
||||
# if fail_any: return False
|
||||
# else:
|
||||
# log.info(f'No Address ID found.')
|
||||
# # from app.methods.address_methods import create_address_obj_v3
|
||||
# if create_address_obj_result := create_address_obj(
|
||||
# account_id = account_id,
|
||||
# for_type = 'contact',
|
||||
# for_id = contact_id,
|
||||
# address_obj_new = address_obj_unknown,
|
||||
# fail_any = fail_any,
|
||||
# ):
|
||||
# if isinstance(create_address_obj_result, int):
|
||||
# address_id = create_address_obj_result
|
||||
# log.info(f'Address created. Address ID: {address_id}')
|
||||
# else:
|
||||
# log.warning(f'Address not created. Contact ID: {contact_id}')
|
||||
# log.debug(create_address_obj_result)
|
||||
# address_id = None
|
||||
# if fail_any: return False
|
||||
# else:
|
||||
# log.warning(f'Address not created. Contact ID: {contact_id}')
|
||||
# log.debug(create_address_obj_result)
|
||||
# address_id = None
|
||||
# if fail_any: return False
|
||||
# # return_dict['address_id'] = address_id
|
||||
# else:
|
||||
# log.info('Address not found or not in a dict.')
|
||||
# pass
|
||||
|
||||
|
||||
|
||||
@@ -521,15 +615,17 @@ def update_contact_obj(
|
||||
|
||||
|
||||
|
||||
contact_dict_up = contact_obj_up.dict(by_alias=False, exclude_unset=True, exclude={'address'})
|
||||
log.debug(contact_dict_up)
|
||||
# contact_dict_up = contact_obj_up.dict(by_alias=False, exclude_unset=True, exclude={'address'})
|
||||
# log.debug(contact_dict_up)
|
||||
|
||||
if contact_obj_up_result := sql_update(data=contact_dict_up, table_name='contact', rm_id_random=True):
|
||||
log.debug(contact_obj_up_result)
|
||||
return True
|
||||
else:
|
||||
log.debug(contact_obj_up_result)
|
||||
return False
|
||||
# if contact_obj_up_result := sql_update(data=contact_dict_up, table_name='contact', rm_id_random=True):
|
||||
# log.debug(contact_obj_up_result)
|
||||
# return True
|
||||
# else:
|
||||
# log.debug(contact_obj_up_result)
|
||||
# return False
|
||||
|
||||
return True
|
||||
# ### END ### API Contact Methods ### update_contact_obj() ###
|
||||
|
||||
|
||||
|
||||
Reference in New Issue
Block a user