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

Code clean up. Bug fixes for person, user, contact, and address methods

Browse Source
This commit is contained in:
Scott Idem
2022-01-06 16:56:38 -05:00
parent c01e668d9e
commit 597c765673
6 changed files with 130 additions and 58 deletions
Download Patch File Download Diff File Expand all files Collapse all files

26
app/methods/contact_methods.py
View File

@@ -319,6 +319,7 @@ def create_contact_obj(
contact_dict_obj: Contact_Base,
for_type: str = None,
for_id: int|str = None,
address_id: int = None,
create_sub_obj: bool = False,
fail_any: bool = False, # Fail if any thing goes wrong for sub objects
) -> int|bool:
@@ -333,6 +334,10 @@ def create_contact_obj(
if for_id := redis_lookup_id_random(record_id_random=for_id, table_name=for_type): pass
else: return False
if address_id := redis_lookup_id_random(record_id_random=address_id, table_name='address'): pass
elif address_id is None: pass
else: return False
log.info('Create dictionary or Pydantic object')
log.debug(type(contact_dict_obj))
if isinstance(contact_dict_obj, dict):
@@ -382,6 +387,8 @@ def create_contact_obj(
contact_dict['for_type'] = for_type
contact_dict['for_id'] = for_id
# if not address_id and contact_obj.address_id_random:
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
@@ -389,7 +396,7 @@ def create_contact_obj(
log.debug(contact_dict_in_result)
contact_id = contact_dict_in_result
if address_id and contact.address:
if address_id and contact_obj.address:
log.info('Updating Address object')
if address_update_result := update_address_obj(
address_id = address_id,
@@ -398,7 +405,7 @@ def create_contact_obj(
for_id = contact_id,
): pass
else: return False
elif contact.address:
elif contact_obj.address:
log.info('Creating Address object')
if address_create_result := create_address_obj(
account_id = account_id,
@@ -422,7 +429,7 @@ def create_contact_obj(
# )
# 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.
# # NOTE: This last update should no longer be needed now that the contact_obj.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
@@ -447,6 +454,7 @@ def create_contact_obj(
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_dict_obj: Contact_Base,
address_id: int = None,
create_sub_obj: bool = False,
fail_any: bool = False, # Fail if any thing goes wrong for sub objects
return_dict: bool = False,
@@ -459,6 +467,10 @@ def update_contact_obj(
if contact_id := redis_lookup_id_random(record_id_random=contact_id, table_name='contact'): pass
else: return False
if address_id := redis_lookup_id_random(record_id_random=address_id, table_name='address'): pass
elif address_id is None: pass
else: return False
log.info('Create dictionary or Pydantic object')
log.debug(type(contact_dict_obj))
if isinstance(contact_dict_obj, dict):
@@ -488,7 +500,7 @@ def update_contact_obj(
log.debug(contact_dict_up_result)
if address_id and contact.address:
if address_id and contact_obj.address:
log.info('Updating Address object')
if address_update_result := update_address_obj(
address_id = address_id,
@@ -497,7 +509,7 @@ def update_contact_obj(
for_id = contact_id,
): pass
else: return False
elif contact.address:
elif contact_obj.address:
log.info('Creating Address object')
if address_create_result := create_address_obj(
account_id = account_id,
@@ -593,7 +605,7 @@ def update_contact_obj(
# log.debug(address_obj_up_result)
# return False
# elif contact_obj_up.address and not contact_obj_up.address.id:
# # NOTE: This will blindly create a new address even if there was one associated but the contact.address_id was not found.
# # NOTE: This will blindly create a new address even if there was one associated but the contact_obj.address_id was not found.
# address_obj_in = contact_obj_up.address
# log.debug(address_obj_in)
# if address_obj_in_result := create_address_obj(
@@ -604,7 +616,7 @@ def update_contact_obj(
# ):
# # log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
# log.debug(address_obj_in_result)
# # NOTE: This last update should no longer be needed now that the contact.address_id is not supposed to be used.
# # NOTE: This last update should no longer be needed now that the contact_obj.address_id is not supposed to be used.
# # Need to update the contact with the new address_id
# contact_obj_up.address_id = address_obj_in_result # REMOVE
# else:

41
app/methods/person_methods.py
View File

@@ -1,5 +1,5 @@
from __future__ import annotations
import datetime, pytz
import datetime, pytz, secrets
from typing import Dict, List, Optional, Set, Union
from pydantic import BaseModel, EmailStr, Field, PrivateAttr, ValidationError, validator
@@ -12,7 +12,7 @@ from app.methods.contact_methods import create_contact_obj, create_update_contac
from app.methods.order_cart_methods import get_order_cart_id_for_person_id, load_order_cart_obj
from app.methods.order_methods import get_order_rec_list, load_order_obj
from app.methods.organization_methods import create_update_organization_obj, load_organization_obj, update_organization_obj
# from app.methods.user_methods import create_user_obj # , load_user_obj, update_user_obj
# from app.methods.user_methods import create_user_obj, load_user_obj, update_user_obj
from app.models.common_field_schema import default_num_bytes
from app.models.contact_models import Contact_Base
@@ -517,6 +517,7 @@ def create_person_kiss(
if user_id and person_obj.user:
log.info('Updating User object')
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,
@@ -525,10 +526,12 @@ def create_person_kiss(
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
@@ -608,6 +611,7 @@ def update_person_kiss(
if user_id and person_obj.user:
log.info('Updating User object')
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,
@@ -616,10 +620,12 @@ def update_person_kiss(
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
@@ -1695,10 +1701,10 @@ def update_person_obj(
# # ### END ### API Person Methods ### create_update_person_obj_v4b() ###
# ### BEGIN ### Person Methods ### email_person_create_url() ###
# ### BEGIN ### Person Methods ### handle_email_person_auth_key_url() ###
# This emails the actual one time use account creation URL for a person.
# Updated 2021-12-03
def email_person_create_url(
def handle_email_person_auth_key_url(
account_id: int|str,
person_id: int|str,
root_url: str,
@@ -1716,6 +1722,23 @@ def email_person_create_url(
person_id = person_id,
):
log.info('Person object loaded')
if person_obj.enable and person_obj.allow_auth_key:
new_auth_key = secrets.token_urlsafe(default_num_bytes)
update_person_data = {}
update_person_data['auth_key'] = new_auth_key
if person_rec_update_result := sql_update(
table_name = 'person',
record_id = person_id,
data = update_person_data
):
log.info('The person record was updated with a new auth_key')
else:
log.warning('The person record was not updated with a new auth_key')
return False
else:
log.warning('The person record was not enabled or auth_key is not allowed')
return False
else: return False
log.debug(person_obj)
@@ -1743,9 +1766,9 @@ def email_person_create_url(
account_short_name = account_cfg.account_short_name
# person_create_url = f'{root_url}person/{person_id_random}/create'
# person_create_auth_key_url = f'{root_url}?user_id={user_id_random}&auth_key={new_auth_key}'
person_create_auth_key_url = f'{root_url}?person_id={user_id_random}&auth_key={new_auth_key}'
# person_auth_key_url = f'{root_url}person/{person_id_random}/create'
# person_auth_key_url = f'{root_url}?user_id={user_id_random}&auth_key={new_auth_key}'
person_auth_key_url = f'{root_url}?person_id={person_id_random}&auth_key={new_auth_key}'
subject = f'{account_short_name}: One Time Use Create Account Link'
# subject = f'{account_short_name}: One Time Use Create Account Link ({new_auth_key})'
@@ -1757,10 +1780,10 @@ def email_person_create_url(
<p>The link below can only be used once. If you would like try again using this method, you must <a href="NOT READY YET">request a new account creation link</a>. If you request multiple links, only the newest link will work.</p>
<p><strong><a href="{person_create_url}" style="appearance: button; display: inline-block; text-align: center; text-decoration: none; padding: .2rem .4rem; border: solid thin gray; border-radius: .2rem; background-color: lightyellow; color: black; font-size: larger;">Click to Finish Account Creation With One Time Use Link</a></strong></p>
<p><strong><a href="{person_auth_key_url}" style="appearance: button; display: inline-block; text-align: center; text-decoration: none; padding: .2rem .4rem; border: solid thin gray; border-radius: .2rem; background-color: lightyellow; color: black; font-size: larger;">Click to Finish Account Creation With One Time Use Link</a></strong></p>
<p>Or copy and paste the link:<br>
<strong style="background-color: lightyellow; color: black; font-size: larger;"><a href="{person_create_url}">{person_create_url}</a></strong></p>
<strong style="background-color: lightyellow; color: black; font-size: larger;"><a href="{person_auth_key_url}">{person_auth_key_url}</a></strong></p>
<p>If you have questions about this email or trouble with this one time use link, you can email <a href="mailto:{help_tech_email}">{help_tech_name} ({help_tech_email})</a>.</p>

102
app/methods/user_methods.py
View File

@@ -6,14 +6,14 @@ from typing import Dict, List, Optional, Set, Union
from pydantic import BaseModel, EmailStr, Field, PrivateAttr, ValidationError, validator
from app.db_sql import redis_lookup_id_random, sql_insert, sql_select, sql_update
from app.lib_general import log, logging, logger_reset, send_email
from app.lib_general import log, logging, logger_reset, secure_hash_string, send_email
# from app.methods.account_methods import load_account_cfg_obj
from app.methods.contact_methods import load_contact_obj, update_contact_obj
# from app.methods.contact_methods import create_contact_obj, load_contact_obj, update_contact_obj
from app.methods.order_methods import load_order_obj, get_order_rec_list
from app.methods.organization_methods import load_organization_obj, update_organization_obj
from app.methods.organization_methods import load_organization_obj # , update_organization_obj
from app.methods.person_methods import create_person_obj_v3, load_person_obj, update_person_obj
from app.methods.post_methods import get_post_rec_list, load_post_obj
# from app.methods.post_methods import get_post_rec_list, load_post_obj
from app.methods.user_role_methods import get_user_role_rec_list, load_user_role_obj
from app.models.common_field_schema import default_num_bytes
@@ -34,7 +34,7 @@ def create_user_obj(
fail_any: bool = True, # Fail if any thing goes wrong for sub objects
return_dict: bool = False,
) -> bool|dict|int:
log.setLevel(logging.INFO) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
# ### SECTION ### Secondary data validation
@@ -42,6 +42,10 @@ def create_user_obj(
if account_id := redis_lookup_id_random(record_id_random=account_id, table_name='account'): pass
else: return False
if person_id := redis_lookup_id_random(record_id_random=person_id, table_name='person'): pass
elif person_id is None: pass
else: return False
log.info('Create dictionary or Pydantic object')
log.debug(type(user_dict_obj))
if isinstance(user_dict_obj, dict):
@@ -56,28 +60,31 @@ def create_user_obj(
user_obj = user_dict_obj
user_obj.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'})
# 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
# 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
user_dict = user_obj.dict(by_alias=False, exclude_defaults=False, exclude_unset=True, exclude={'contact', 'contact_id_random', 'new_password', 'organization', 'person', 'person_id_random', 'created_on', 'updated_on'})
# ### 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?
if user_obj.new_password:
log.debug(user_obj.new_password)
else:
user_obj.new_password = secrets.token_urlsafe(default_num_bytes)
hash_string = secure_hash_string(string=user_obj.new_password)
user_obj.password = hash_string
user_dict['password'] = hash_string
log.debug(user_obj.new_password)
# 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?
if person_id:
# Link to an existing person
log.info(f'Adding person_id to user_dict. User ID: {person_id}')
user_obj.person_id = person_id # Is this needed?
user_dict['person_id'] = person_id
log.debug(user_obj)
log.debug(user_dict)
# if account_id := redis_lookup_id_random(record_id_random=account_id, table_name='account'): pass
@@ -87,7 +94,7 @@ def create_user_obj(
# 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}')
log.info(f'Checking if the username is already in use for the account... Account: {account_id}; Username: {username}')
sql_select_user = f"""
SELECT user.id, user.id_random, user.name, user.email
FROM `user` AS user
@@ -96,23 +103,27 @@ def create_user_obj(
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}')
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_user_result.get('id', None)
log.info('A user account already exists with this username. Current User ID: {user_id} Username: {username}')
person_id_for_user_id = sql_select_user_result.get('person_id', None)
log.info(f'A user account already exists with this username. Current User ID: {user_id}; Username: {username}; Person ID for User: {person_id_for_user_id}; Person ID: {person_id}')
if allow_update:
log.info('Updating instead of inserting. Current User ID: {user_id} Username: {username}')
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_dict_up_result)
return False
log.info(f'Updating instead of inserting. Current User ID: {user_id}; Username: {username}')
# NOTE: Should this call the update_user_obj() function instead??? NOTE NOTE NOTE NOTE
if person_id_for_user_id == person_id or person_id_for_user_id is None:
user_dict['id'] = user_id
user_dict['person_id'] = person_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_dict_up_result)
return False
else:
log.info('Updating is now allowed. Current User ID: {user_id} Username: {username}')
log.info(f'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}')
log.info(f'Avoiding duplicate username is now allowed. Suggested Username: {username}')
new_username = username+'-'+str(random.randint(10, 99))
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
@@ -160,6 +171,10 @@ def update_user_obj(
if user_id := redis_lookup_id_random(record_id_random=user_id, table_name='user'): pass
else: return False
if person_id := redis_lookup_id_random(record_id_random=person_id, table_name='person'): pass
elif person_id is None: pass
else: return False
log.info('Create dictionary or Pydantic object')
log.debug(type(user_dict_obj))
if isinstance(user_dict_obj, dict):
@@ -209,7 +224,7 @@ def update_user_obj(
# log.debug(contact_obj_up_result)
# return False
# 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.
# # NOTE: This will blindly create a new contact even if there was one associated but the user_obj.contact_id was not found.
# contact_obj_in = user_obj.contact
# log.debug(contact_obj_in)
# if contact_obj_in_result := create_contact_obj(contact_dict_obj=contact_obj_in):
@@ -256,7 +271,7 @@ def update_user_obj(
# log.debug(person_obj_up_result)
# return False
# 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.
# # NOTE: This will blindly create a new person even if there was one associated but the user_obj.person_id was not found.
# 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):
@@ -276,12 +291,25 @@ def update_user_obj(
user_obj.id = user_id # Is this needed?
user_dict['id'] = user_id
if user_obj.new_password:
log.debug(user_obj.new_password)
else:
user_obj.new_password = secrets.token_urlsafe(default_num_bytes)
hash_string = secure_hash_string(string=user_obj.new_password)
user_obj.password = hash_string
user_dict['password'] = hash_string
log.debug(user_obj.new_password)
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
log.debug(user_obj)
log.debug(user_dict)
if user_dict_up_result := sql_update(data=user_dict, table_name='user', rm_id_random=True): pass
else:
log.warning(f'User not updated.')

11
app/models/user_models.py
View File

@@ -131,7 +131,7 @@ class User_New_Base(BaseModel):
log.setLevel(logging.WARNING)
log.debug(locals())
if values['new_password']:
if values.get('new_password'):
return secure_hash_string(string=values['new_password'])
return None
@@ -361,6 +361,15 @@ class User_Base(BaseModel):
return redis_lookup_id_random(record_id_random=values['person_id_random'], table_name='person')
return None
@validator('password', always=True)
def hash_new_password(cls, v, values, **kwargs):
log.setLevel(logging.WARNING)
log.debug(locals())
if values.get('new_password'):
return secure_hash_string(string=values['new_password'])
return None
class Config:
underscore_attrs_are_private = True
allow_population_by_field_name = True

6
app/routers/person.py
View File

@@ -10,7 +10,7 @@ from app.db_sql import sql_insert, sql_update, sql_insert_or_update, sql_select,
from app.routers.api_crud import delete_obj_template, get_obj_template, get_obj_li_template, patch_obj_template, post_obj_template
# from app.methods.membership_person_methods import load_membership_person_obj
from app.methods.person_methods import create_person_kiss, create_update_person_obj_v4b, email_person_create_url, get_person_rec_list, get_person_rec_w_external_id, load_person_obj, update_person_obj, update_person_kiss
from app.methods.person_methods import create_person_kiss, create_update_person_obj_v4b, handle_email_person_auth_key_url, get_person_rec_list, get_person_rec_w_external_id, load_person_obj, update_person_obj, update_person_kiss
from app.models.person_models import Person_Base
from app.models.response_models import Resp_Body_Base, mk_resp
@@ -460,7 +460,7 @@ async def lookup_email(
# @router.get('/person/{person_id}/email_create_url', response_model=Resp_Body_Base)
@router.get('/person/{person_id}/email_auth_key_url', response_model=Resp_Body_Base)
async def email_auth_key_url(
person_id: Optional[str] = Query(None, min_length=11, max_length=22),
person_id: str = Query(None, min_length=11, max_length=22),
root_url: Optional[str] = Query(None, min_length=10, max_length=100), # Absolute min = 7
commons: Common_Route_Params = Depends(common_route_params),
):
@@ -472,7 +472,7 @@ async def email_auth_key_url(
if person_id := redis_lookup_id_random(record_id_random=person_id, table_name='person'): pass
else: return mk_resp(data=False, status_code=404, response=commons.response) # Not Found
if result := email_person_create_url(
if result := handle_email_person_auth_key_url(
account_id = account_id,
person_id = person_id,
root_url = root_url,

2
app/routers/user.py
View File

@@ -775,7 +775,7 @@ async def lookup_username(
# @router.get('/user/email_auth_key_url', response_model=Resp_Body_Base)
@router.get('/user/{user_id}/email_auth_key_url', response_model=Resp_Body_Base)
async def email_auth_key_url(
user_id: Optional[str] = Query(None, min_length=11, max_length=22),
user_id: str = Query(..., min_length=11, max_length=22),
root_url: Optional[str] = Query(None, min_length=10, max_length=100), # Absolute min = 7
return_obj: bool = False,
commons: Common_Route_Params = Depends(common_route_params),