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

Code clean up. Person and User related is being worked on.

Browse Source
This commit is contained in:
Scott Idem
2022-01-06 12:25:53 -05:00
parent a2de9572ba
commit c127e0822c
10 changed files with 166 additions and 91 deletions
Download Patch File Download Diff File Expand all files Collapse all files

181
app/methods/user_methods.py
View File

@@ -28,7 +28,8 @@ from app.models.user_models import User_Base, User_New_Base, User_Out_Base
# Reviewed and updated 2021-08-10
def create_user_obj(
account_id: int|str,
user_obj_new: User_New_Base,
user_dict_obj: User_New_Base,
person_id: int,
allow_update: bool = False, # Allow updating the user account if one is found
avoid_dup_username: bool = False, # Avoid creating a duplicate by modifying the supplied username
create_sub_obj: bool = False,
@@ -38,33 +39,55 @@ def create_user_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 account_id := redis_lookup_id_random(record_id_random=account_id, table_name='account'): pass
else: return False
log.debug(type(user_obj_new))
if isinstance(user_obj_new, dict):
user_obj_new['account_id'] = account_id
log.info('Create dictionary or Pydantic object')
log.debug(type(user_dict_obj))
if isinstance(user_dict_obj, dict):
user_dict = user_dict_obj
try:
user_obj_new = User_New_Base(**user_obj_new)
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(user_obj_new)
user_obj = Person_Base(**user_dict)
log.debug(user_obj)
except ValidationError as e:
log.error(e.json())
return False
else:
user_obj = user_dict_obj
user_obj.account_id = account_id
user_obj_data = user_obj_new.dict(by_alias=False, exclude_defaults=False, exclude_unset=True, exclude={'contact', 'new_password', 'organization', 'person', 'created_on', 'updated_on'})
log.debug(user_obj_data)
user_obj_data['account_id'] = account_id
user_dict = user_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', 'new_password', 'organization', 'person', 'user', 'created_on', 'updated_on'})
user_obj_data['password'] = user_obj_new.password # There has to be a better way to do this??? It thinks "password" is unset and so is excluded?
log.debug(user_obj_data)
# log.debug(type(user_dict_obj))
# if isinstance(user_dict_obj, dict):
# user_dict_obj['account_id'] = account_id
# try:
# user_obj = User_New_Base(**user_dict_obj)
# log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
# log.debug(user_obj)
# except ValidationError as e:
# log.error(e.json())
# return False
if account_id := redis_lookup_id_random(record_id_random=account_id, table_name='account'): pass
elif account_id := user_obj_data.get('account_id', None): pass
else: return False
# user_dict = user_obj.dict(by_alias=False, exclude_defaults=False, exclude_unset=True, exclude={'contact', 'new_password', 'organization', 'person', 'created_on', 'updated_on'})
# log.debug(user_dict)
# user_dict['account_id'] = account_id
# account_id = user_obj_data.get('account_id', None)
username = user_obj_data.get('username', None)
# ### SECTION ### Process data
user_obj.account_id = account_id # Is this needed?
user_dict['account_id'] = account_id
user_dict['password'] = user_obj.password # There has to be a better way to do this??? It thinks "password" is unset and so is excluded?
log.debug(user_dict)
# if account_id := redis_lookup_id_random(record_id_random=account_id, table_name='account'): pass
# elif account_id := user_dict.get('account_id', None): pass
# else: return False
# account_id = user_dict.get('account_id', None)
username = user_dict.get('username', None)
log.info(f'Checking if the username is already in use for the account... Account: {account_id} Username: {username}')
sql_select_user = f"""
@@ -73,58 +96,60 @@ def create_user_obj(
WHERE user.account_id = :account_id and user.username = :username
"""
if sql_select_result := sql_select(sql=sql_select_user, data=user_obj_data, rm_id_random=True):
if isinstance(sql_select_result, list):
if sql_select_user_result := sql_select(sql=sql_select_user, data=user_dict, rm_id_random=True):
if isinstance(sql_select_user_result, list):
log.exception(f'Multiple user accounts already exists with this username. The database needs to be checked. Account ID: {account_id} Username: {username}')
return False
user_id = sql_select_result.get('id', None)
user_id = sql_select_user_result.get('id', None)
log.info('A user account already exists with this username. Current User ID: {user_id} Username: {username}')
if allow_update:
log.info('Updating instead of inserting. Current User ID: {user_id} Username: {username}')
user_obj_data['id'] = user_id
if user_obj_up_result := sql_update(data=user_obj_data, table_name='user', rm_id_random=True):
user_dict['id'] = user_id
if user_dict_up_result := sql_update(data=user_dict, table_name='user', rm_id_random=True):
log.info(f'User updated with new user data. User ID: {user_id}')
else:
log.warning(f'User not updated with new user data. User ID: {user_id}')
log.debug(user_obj_up_result)
log.debug(user_dict_up_result)
return False
else:
log.info('Updating is now allowed. Current User ID: {user_id} Username: {username}')
if avoid_dup_username:
log.info('Avoiding duplicate username is now allowed. Suggested Username: {username}')
new_username = username+'-'+str(random.randint(10, 99))
user_obj_data['username'] = new_username
if user_obj_in_result := sql_insert(data=user_obj_data, table_name='user', rm_id_random=True, id_random_length=8): pass
user_dict['username'] = new_username
if user_dict_in_result := sql_insert(data=user_dict, table_name='user', rm_id_random=True, id_random_length=8): pass
else:
log.warning(f'User not created.')
log.debug(user_obj_in_result)
log.debug(user_dict_in_result)
return False
user_id = user_obj_in_result
user_id = user_dict_in_result
else:
log.warning(f'Updating is not allowed and avoid duplicate username is not allowed. Username: {username}')
return False
#log.setLevel(logging.DEBUG)
# log.debug(user_obj_up_result)
# log.debug(user_dict_up_result)
log.debug(f'Returning the existing user_id: {user_id}')
else:
if user_obj_in_result := sql_insert(data=user_obj_data, table_name='user', rm_id_random=True, id_random_length=8): pass
if user_dict_in_result := sql_insert(data=user_dict, table_name='user', rm_id_random=True, id_random_length=8): pass
else:
log.warning(f'User not created.')
log.debug(user_obj_in_result)
log.debug(user_dict_in_result)
return False
user_id = user_obj_in_result
user_id = user_dict_in_result
log.info(f'Returning the new user_id: {user_id}')
log.debug(f'Returning the new user_id: {user_id}')
return user_id
# ### END ### API User Methods ### create_user_obj() ###
# ### BEGIN ### API User Methods ### update_user_obj() ###
# Updated 2021-08-25
# Updated 2022-01-06
def update_user_obj(
user_id: int|str, # Ideally the int ID should be passed. This allows for updating of the id_random value.
user_obj_up: User_Base,
user_dict_obj: User_Base,
person_id: int,
create_sub_obj: bool = False,
fail_any: bool = True, # Fail if any thing goes wrong for sub objects
return_dict: bool = False,
@@ -132,29 +157,49 @@ def update_user_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 user_id := redis_lookup_id_random(record_id_random=user_id, table_name='user'): pass
else: return False
log.debug(type(user_obj_up))
if isinstance(user_obj_up, dict):
log.info('Create dictionary or Pydantic object')
log.debug(type(user_dict_obj))
if isinstance(user_dict_obj, dict):
user_dict = user_dict_obj
try:
user_obj_up = User_Base(**user_obj_up)
user_obj = Person_Base(**user_dict)
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(user_obj_up)
log.debug(user_obj)
except ValidationError as e:
log.error(e.json())
return False
else:
user_obj = user_dict_obj
# IMPORTANT NOTE: Need to be extra careful if allowing an update to password, super, or manager. Maybe other fields?
user_dict = user_obj.dict(by_alias=False, exclude_unset=True, exclude={'password', 'super', 'manager', 'contact', 'organization', 'person', 'created_on', 'updated_on'})
log.debug(type(user_dict_obj))
if isinstance(user_dict_obj, dict):
try:
user_obj = User_Base(**user_dict_obj)
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(user_obj)
except ValidationError as e:
log.error(e.json())
return False
user_obj_up.id = user_id
user_obj.id = user_id
log.debug(user_obj_up)
# log.debug(user_obj_up.dict(by_alias=True, exclude_unset=True))
log.debug(user_obj_up.dict(by_alias=False, exclude_unset=True))
# log.debug(user_obj_up.dict(by_alias=False, exclude_unset=False))
log.debug(user_obj)
# log.debug(user_obj.dict(by_alias=True, exclude_unset=True))
log.debug(user_obj.dict(by_alias=False, exclude_unset=True))
# log.debug(user_obj.dict(by_alias=False, exclude_unset=False))
# if user_obj_up.contact_id and user_obj_up.contact:
# contact_id = user_obj_up.contact_id
# contact_obj_up = user_obj_up.contact
# if user_obj.contact_id and user_obj.contact:
# contact_id = user_obj.contact_id
# contact_obj_up = user_obj.contact
# log.debug(contact_id)
# log.debug(contact_obj_up)
# if contact_obj_up_result := update_contact_obj(
@@ -166,20 +211,20 @@ def update_user_obj(
# else:
# log.debug(contact_obj_up_result)
# return False
# elif user_obj_up.contact and not user_obj_up.contact.id:
# elif user_obj.contact and not user_obj.contact.id:
# # NOTE: This will blindly create a new contact even if there was one associated but the user.contact_id was not found.
# contact_obj_in = user_obj_up.contact
# contact_obj_in = user_obj.contact
# log.debug(contact_obj_in)
# if contact_obj_in_result := create_contact_obj(contact_obj_new=contact_obj_in):
# # log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
# log.debug(contact_obj_in_result)
# user_obj_up.contact_id = contact_obj_in_result
# user_obj.contact_id = contact_obj_in_result
# else:
# # log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
# log.debug(contact_obj_in_result)
# return False
# if organization_obj_update := user_obj_up.organization:
# if organization_obj_update := user_obj.organization:
# log.debug(organization_obj_update)
# if organization_obj_up_result := update_organization_obj(organization_obj_update):
# log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
@@ -199,9 +244,9 @@ def update_user_obj(
# log.debug(organization_obj_in_result)
# return False
# if user_obj_up.person_id and user_obj_up.person:
# person_id = user_obj_up.person_id
# person_obj_up = user_obj_up.person
# if user_obj.person_id and user_obj.person:
# person_id = user_obj.person_id
# person_obj_up = user_obj.person
# log.debug(person_id)
# log.debug(person_obj_up)
# if person_obj_up_result := update_person_obj(
@@ -213,9 +258,9 @@ def update_user_obj(
# else:
# log.debug(person_obj_up_result)
# return False
# elif user_obj_up.person and not user_obj_up.person.id:
# elif user_obj.person and not user_obj.person.id:
# # NOTE: This will blindly create a new person even if there was one associated but the user.person_id was not found.
# person_obj_in = user_obj_up.person
# person_obj_in = user_obj.person
# log.debug(person_obj_in)
# if person_obj_in_result := create_person_obj_v3(person_obj_new=person_obj_in):
# # log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
@@ -227,16 +272,28 @@ def update_user_obj(
# return False
# IMPORTANT NOTE: Need to be extra careful if allowing an update to password, super, or manager. Maybe other fields?
user_dict_up = user_obj_up.dict(by_alias=False, exclude_unset=True, exclude={'password', 'super', 'manager', 'contact', 'organization', 'person'})
log.debug(user_dict_up)
# user_dict = user_obj.dict(by_alias=False, exclude_unset=True, exclude={'password', 'super', 'manager', 'contact', 'organization', 'person'})
# log.debug(user_dict)
if user_obj_up_result := sql_update(data=user_dict_up, table_name='user', rm_id_random=True):
log.debug(user_obj_up_result)
return True
# return user_id
# ### SECTION ### Process data
user_obj.user_id = user_id # Is this needed?
user_dict['user_id'] = user_id
if person_id:
# Link to an existing person
log.info(f'Adding person_id to person_dict. Person ID: {person_id}')
user_obj.person_id = person_id # Is this needed?
user_dict['person_id'] = person_id
if user_dict_up_result := sql_update(data=user_dict, table_name='user', rm_id_random=True): pass
else:
log.debug(user_obj_up_result)
log.warning(f'User not updated.')
log.debug(user_dict_up_result)
return False
log.debug(user_dict_up_result)
return True
# ### END ### API User Methods ### update_user_obj() ###