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

Browse Source
This commit is contained in:
Scott Idem
2022-01-18 20:03:46 -05:00
parent d6caa819b0
commit 3d2c7282dd
10 changed files with 722 additions and 440 deletions
Download Patch File Download Diff File Expand all files Collapse all files

445
app/methods/person_methods.py
View File

@@ -19,6 +19,229 @@ from app.models.contact_models import Contact_Base
from app.models.person_models import Person_Base
# ### BEGIN ### API Person Methods ### create_person_kiss() ###
# Updated 2022-01-06
def create_person_kiss(
account_id: int,
person_dict_obj: Person_Base,
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())
# ### SECTION ### Secondary data validation
log.info('Create dictionary or Pydantic object')
log.debug(type(person_dict_obj))
if isinstance(person_dict_obj, dict):
person_dict = person_dict_obj
try:
person_obj = Person_Base(**person_dict)
log.debug(person_obj)
except ValidationError as e:
log.error(e.json())
return False
else:
person_obj = person_dict_obj
# 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 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
# Look for a contact_id in the contact_obj
if contact_id: pass
elif contact_id := person_obj.contact.id: pass
# Look for a user_id in the person_obj
if user_id: pass
elif user_id := person_obj.user.id: pass
# if user_id:
# # Link to an existing user
# log.info(f'Adding user_id to person_dict. User ID: {user_id}')
# 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
else:
log.warning(f'Person not created.')
log.debug(person_dict_in_result)
return False
log.debug(person_dict_in_result)
person_id = person_dict_in_result
if contact_id and person_obj.contact:
log.info('Updating Contact object')
if contact_update_result := update_contact_obj(
contact_id = contact_id,
contact_dict_obj = person_obj.contact,
for_type = 'person',
for_id = person_id,
): pass
else: return False
elif person_obj.contact:
log.info('Creating Contact object')
if contact_create_result := create_contact_obj(
account_id = account_id,
contact_dict_obj = person_obj.contact,
for_type = 'person',
for_id = person_id,
): pass
else: return False
else: pass
if user_id and person_obj.user:
log.info('Updating User object')
# Link to an existing user
log.info(f'Adding user_id to person_dict. User ID: {user_id}')
person_obj.user_id = user_id # Is this needed?
person_dict['user_id'] = user_id
from app.methods.user_methods import update_user_obj # NOTE: This creates a loop if outside function
if user_update_result := update_user_obj(
user_id = user_id,
user_dict_obj = person_obj.user,
person_id = person_id,
): pass # NOTE: There is a trigger that will update the person record with the new user ID.
else: return False
elif person_obj.user:
log.info('Creating User object')
from app.methods.user_methods import create_user_obj # NOTE: This creates a loop if outside function
if user_create_result := create_user_obj(
account_id = account_id,
user_dict_obj = person_obj.user,
person_id = person_id,
allow_update = True, # WARNING NOTE: This will allow an existing user record to be updated.
): pass # NOTE: There is a trigger that will update the person record with the new user ID.
else: return False
else: pass
log.info(f'Returning the Person ID: {person_id}')
return person_id
# ### END ### API Person Methods ### create_person_kiss() ###
# ### BEGIN ### API Person Methods ### update_person_kiss() ###
# Updated 2022-01-06
def update_person_kiss(
person_id: int,
person_dict_obj: Person_Base,
contact_id: int|None = None,
organization_id: int|None = None,
user_id: int|None = None,
):
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
# ### SECTION ### Secondary data validation
log.info('Create dictionary or Pydantic object')
log.debug(type(person_dict_obj))
if isinstance(person_dict_obj, dict):
person_dict = person_dict_obj
try:
person_obj = Person_Base(**person_dict)
log.debug(person_obj)
except ValidationError as e:
log.error(e.json())
return False
else:
person_obj = person_dict_obj
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
person_obj.id = person_id # Is this needed?
person_dict['id'] = person_id
# Look for a contact_id in the contact_obj
if contact_id: pass
elif contact_id := person_obj.contact.id: pass
# Look for a user_id in the person_obj
if user_id: pass
elif user_id := person_obj.user.id: pass
# if user_id:
# # Link to an existing user
# log.info(f'Adding user_id to person_dict. User ID: {user_id}')
# 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,
): pass
else:
log.warning(f'Person not updated.')
log.debug(person_dict_up_result)
return False
log.debug(person_dict_up_result)
if contact_id and person_obj.contact:
log.info('Updating Contact object')
if contact_update_result := update_contact_obj(
contact_id = contact_id,
contact_dict_obj = person_obj.contact,
for_type = 'person',
for_id = person_id,
): pass
else: return False
elif person_obj.contact:
log.info('Creating Contact object')
if contact_create_result := create_contact_obj(
account_id = account_id,
contact_dict_obj = person_obj.contact,
for_type = 'person',
for_id = person_id,
): pass
else: return False
else: pass
if user_id and person_obj.user:
log.info('Updating User object')
# Link to an existing user
log.info(f'Adding user_id to person_dict. User ID: {user_id}')
person_obj.user_id = user_id # Is this needed?
person_dict['user_id'] = user_id
from app.methods.user_methods import update_user_obj # NOTE: This creates a loop if outside function
if user_update_result := update_user_obj(
user_id = user_id,
user_dict_obj = person_obj.user,
person_id = person_id,
): pass # NOTE: There is a trigger that will update the person record with the new user ID.
else: return False
elif person_obj.user:
log.info('Creating User object')
from app.methods.user_methods import create_user_obj # NOTE: This creates a loop if outside function
if user_create_result := create_user_obj(
account_id = account_id,
user_dict_obj = person_obj.user,
person_id = person_id,
allow_update = True, # WARNING NOTE: This will allow an existing user record to be updated.
): pass # NOTE: There is a trigger that will update the person record with the new user ID.
else: return False
else: pass
return True
# ### END ### API Person Methods ### update_person_kiss() ###
# ### BEGIN ### API Person Methods ### load_person_obj() ###
# Updated 2021-12-15
def load_person_obj(
@@ -430,228 +653,6 @@ def get_person_rec_list(
# data_dict = data_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'])
# ### BEGIN ### API Person Methods ### create_person_kiss() ###
# Updated 2022-01-06
def create_person_kiss(
account_id: int,
person_dict_obj: Person_Base,
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())
# ### SECTION ### Secondary data validation
log.info('Create dictionary or Pydantic object')
log.debug(type(person_dict_obj))
if isinstance(person_dict_obj, dict):
person_dict = person_dict_obj
try:
person_obj = Person_Base(**person_dict)
log.debug(person_obj)
except ValidationError as e:
log.error(e.json())
return False
else:
person_obj = person_dict_obj
# 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 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
# Look for a contact_id in the contact_obj
if contact_id: pass
elif contact_id := person_obj.contact.id: pass
# Look for a user_id in the person_obj
if user_id: pass
elif user_id := person_obj.user.id: pass
# if user_id:
# # Link to an existing user
# log.info(f'Adding user_id to person_dict. User ID: {user_id}')
# 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
else:
log.warning(f'Person not created.')
log.debug(person_dict_in_result)
return False
log.debug(person_dict_in_result)
person_id = person_dict_in_result
if contact_id and person_obj.contact:
log.info('Updating Contact object')
if contact_update_result := update_contact_obj(
contact_id = contact_id,
contact_dict_obj = person_obj.contact,
for_type = 'person',
for_id = person_id,
): pass
else: return False
elif person_obj.contact:
log.info('Creating Contact object')
if contact_create_result := create_contact_obj(
account_id = account_id,
contact_dict_obj = person_obj.contact,
for_type = 'person',
for_id = person_id,
): pass
else: return False
else: pass
if user_id and person_obj.user:
log.info('Updating User object')
# Link to an existing user
log.info(f'Adding user_id to person_dict. User ID: {user_id}')
person_obj.user_id = user_id # Is this needed?
person_dict['user_id'] = user_id
from app.methods.user_methods import update_user_obj # NOTE: This creates a loop if outside function
if user_update_result := update_user_obj(
user_id = user_id,
user_dict_obj = person_obj.user,
person_id = person_id,
): pass # NOTE: There is a trigger that will update the person record with the new user ID.
else: return False
elif person_obj.user:
log.info('Creating User object')
from app.methods.user_methods import create_user_obj # NOTE: This creates a loop if outside function
if user_create_result := create_user_obj(
account_id = account_id,
user_dict_obj = person_obj.user,
person_id = person_id,
allow_update = True, # WARNING NOTE: This will allow an existing user record to be updated.
): pass # NOTE: There is a trigger that will update the person record with the new user ID.
else: return False
else: pass
log.info(f'Returning the Person ID: {person_id}')
return person_id
# ### END ### API Person Methods ### create_person_kiss() ###
# ### BEGIN ### API Person Methods ### update_person_kiss() ###
# Updated 2022-01-06
def update_person_kiss(
person_id: int,
person_dict_obj: Person_Base,
contact_id: int|None = None,
organization_id: int|None = None,
user_id: int|None = None,
):
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
# ### SECTION ### Secondary data validation
log.info('Create dictionary or Pydantic object')
log.debug(type(person_dict_obj))
if isinstance(person_dict_obj, dict):
person_dict = person_dict_obj
try:
person_obj = Person_Base(**person_dict)
log.debug(person_obj)
except ValidationError as e:
log.error(e.json())
return False
else:
person_obj = person_dict_obj
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
person_obj.id = person_id # Is this needed?
person_dict['id'] = person_id
# Look for a contact_id in the contact_obj
if contact_id: pass
elif contact_id := person_obj.contact.id: pass
# Look for a user_id in the person_obj
if user_id: pass
elif user_id := person_obj.user.id: pass
# if user_id:
# # Link to an existing user
# log.info(f'Adding user_id to person_dict. User ID: {user_id}')
# 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
else:
log.warning(f'Person not updated.')
log.debug(person_dict_up_result)
return False
log.debug(person_dict_up_result)
if contact_id and person_obj.contact:
log.info('Updating Contact object')
if contact_update_result := update_contact_obj(
contact_id = contact_id,
contact_dict_obj = person_obj.contact,
for_type = 'person',
for_id = person_id,
): pass
else: return False
elif person_obj.contact:
log.info('Creating Contact object')
if contact_create_result := create_contact_obj(
account_id = account_id,
contact_dict_obj = person_obj.contact,
for_type = 'person',
for_id = person_id,
): pass
else: return False
else: pass
if user_id and person_obj.user:
log.info('Updating User object')
# Link to an existing user
log.info(f'Adding user_id to person_dict. User ID: {user_id}')
person_obj.user_id = user_id # Is this needed?
person_dict['user_id'] = user_id
from app.methods.user_methods import update_user_obj # NOTE: This creates a loop if outside function
if user_update_result := update_user_obj(
user_id = user_id,
user_dict_obj = person_obj.user,
person_id = person_id,
): pass # NOTE: There is a trigger that will update the person record with the new user ID.
else: return False
elif person_obj.user:
log.info('Creating User object')
from app.methods.user_methods import create_user_obj # NOTE: This creates a loop if outside function
if user_create_result := create_user_obj(
account_id = account_id,
user_dict_obj = person_obj.user,
person_id = person_id,
allow_update = True, # WARNING NOTE: This will allow an existing user record to be updated.
): pass # NOTE: There is a trigger that will update the person record with the new user ID.
else: return False
else: pass
return True
# ### END ### API Person Methods ### update_person_kiss() ###
# ### BEGIN ### API Person Methods ### create_update_person_obj_v4b() ###