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

A lot of route common params clean up

Browse Source
This commit is contained in:
Scott Idem
2022-01-05 20:39:34 -05:00
parent 8373f52f22
commit b32900a44f
4 changed files with 238 additions and 34 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
app/lib_general.py
View File

@@ -57,13 +57,14 @@ async def common_route_params(
enabled: str = 'enabled', # all, enabled, disabled
limit: int = 100,
by_alias: bool = True,
exclude: Optional[list] = [],
exclude_none: Optional[bool] = True,
exclude_unset: bool = True,
include: Optional[list] = [],
# NOTE: Uncommenting either exclude or include breaks the JSON body format. I do not know why? Should be: {} Becomes this: {"obj_name": {"data_name": "data_value"}} -STI 2022-01-05
# exclude: Optional[list] = [], # Leaving this and include commented out
# include: Optional[list] = [], # Leaving this and exclude commented out
response: Response = Response,
) -> Common_Route_Params:
log.setLevel(logging.INFO) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
log.info(f'Setting commons values')

169
app/methods/person_methods.py
View File

@@ -424,6 +424,175 @@ def get_person_rec_list(
# ### END ### API Person Methods ### get_person_rec_list() ###
# ### BEGIN ### API Person Methods ### create_person_kiss() ###
# Updated 2022-01-05
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,
):
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
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.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
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'})
if user_id:
# Link to an existing user
log.info(f'Adding user_id to person_dict. User ID: {user_id}')
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.contact:
log.info('Updating Contact object')
if contact_update_result := update_contact(): pass
else: return False
elif person.contact:
log.info('Creating Contact object')
if contact_create_result := create_contact(
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.user:
log.info('Updating User object')
if user_update_result := update_user(): pass
else: return False
elif person.user:
log.info('Creating User object')
if user_create_result := create_user(
account_id = account_id,
user_dict_obj = person_obj.user,
person_id = person_id,
): pass
else: return False
else: pass
return person_id
# ### END ### API Person Methods ### create_person_kiss() ###
# ### BEGIN ### API Person Methods ### update_person_kiss() ###
# Updated 2022-01-05
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())
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.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
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'})
if user_id:
# Link to an existing user
log.info(f'Adding user_id to person_dict. User ID: {user_id}')
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.contact:
log.info('Updating Contact object')
if contact_update_result := update_contact(
contact_dict_obj = person_obj.contact,
for_type = 'person',
for_id = person_id,
): pass
else: return False
elif person.contact:
log.info('Creating Contact object')
if contact_create_result := create_contact(
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.user:
log.info('Updating User object')
if user_update_result := update_user(): pass
else: return False
elif person.user:
log.info('Creating User object')
if user_create_result := create_user(
account_id = account_id,
user_dict_obj = person_obj.user,
person_id = person_id,
):
# user_id = user_create_result
# person_data = {}
# person_data['person_id'] = person_id
# person_data['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)
pass
else: return False
else: pass
return True
# ### END ### API Person Methods ### update_person_kiss() ###
# ### BEGIN ### API Person Methods ### create_update_person_obj_v4b() ###
def create_update_person_obj_v4b(
account_id: int|str,

94
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_update_person_obj_v4b, email_person_create_url, get_person_rec_list, get_person_rec_w_external_id, load_person_obj, update_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.models.person_models import Person_Base
from app.models.response_models import Resp_Body_Base, mk_resp
@@ -19,50 +19,84 @@ from app.models.response_models import Resp_Body_Base, mk_resp
router = APIRouter()
# ### BEGIN ### API Person Routers ### post_person_obj() ###
# Updated 2022-01-05
@router.post('/person', response_model=Resp_Body_Base)
async def post_person_obj(
obj: Person_Base,
return_obj: Optional[bool] = True,
person_obj: Person_Base,
contact_id: str = Query(None, min_length=11, max_length=22),
organization_id: str = Query(None, min_length=11, max_length=22),
user_id: str = Query(None, min_length=11, max_length=22),
return_obj: bool = True,
commons: Common_Route_Params = Depends(common_route_params),
):
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
obj_type = 'person'
obj_data_dict = obj.dict(by_alias=False, exclude_unset=True)
result = post_obj_template(
obj_type=obj_type,
data=obj_data_dict,
return_obj=True,
by_alias=True,
exclude_unset=True,
)
return result
if person_id := create_person_kiss(
account_id = commons.x_account_id,
person_dict_obj = person_obj,
contact_id = contact_id,
organization_id = organization_id,
user_id = user_id,
): pass
else:
log.warning('Likely bad request')
return mk_resp(data=False, status_code=400, response=commons.response) # Bad Request
if return_obj:
person_obj = load_person_obj(
person_id = person_id,
inc_address = False,
inc_contact = False,
inc_organization = False,
inc_user = False
).dict(by_alias=commons.by_alias, exclude_unset=commons.exclude_unset)
data = person_obj
else:
data = person_id
return mk_resp(data=data, response=commons.response)
# ### END ### API Person Routers ### post_person_obj() ###
@router.patch('/person/{obj_id}', response_model=Resp_Body_Base)
# ### BEGIN ### API Person Routers ### patch_person_obj() ###
# Updated 2022-01-05
@router.patch('/person/{person_id}', response_model=Resp_Body_Base)
async def patch_person_obj(
obj: Person_Base,
obj_id: str = Query(..., min_length=11, max_length=22),
person_obj: Person_Base,
person_id: str = Query(..., min_length=11, max_length=22),
contact_id: str = Query(None, min_length=11, max_length=22),
organization_id: str = Query(None, min_length=11, max_length=22),
user_id: str = Query(None, min_length=11, max_length=22),
return_obj: Optional[bool] = True,
commons: Common_Route_Params = Depends(common_route_params),
):
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
obj_type = 'person'
obj_data_dict = obj.dict(by_alias=False, exclude_unset=True)
obj_data_dict['id'] = redis_lookup_id_random(record_id_random=obj_id, table_name=obj_type)
obj_data_dict['id_random'] = obj_id
result = patch_obj_template(
obj_type=obj_type,
data=obj_data_dict,
obj_id=obj_id,
return_obj=True,
by_alias=True,
exclude_unset=True,
)
return result
if person_update_result := update_person_kiss(
person_dict_obj = person_obj,
contact_id = contact_id,
organization_id = organization_id,
user_id = user_id,
): pass
else:
log.warning('Likely bad request')
return mk_resp(data=False, status_code=400, response=commons.response) # Bad Request
if return_obj:
person_obj = load_person_obj(
person_id = person_id,
inc_address = False,
inc_contact = False,
inc_organization = False,
inc_user = False
).dict(by_alias=commons.by_alias, exclude_unset=commons.exclude_unset)
data = person_obj
else:
data = person_id
return mk_resp(data=data, response=commons.response)
# ### END ### API Person Routers ### patch_person_obj() ###
# ### BEGIN ### API Person ### v3_post_person_obj_new() ###

2
app/routers/user.py
View File

@@ -713,7 +713,7 @@ async def lookup_username(
inc_user_role_list: bool = False,
commons: Common_Route_Params = Depends(common_route_params),
):
log.setLevel(logging.INFO) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
account_id = commons.x_account_id