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

General code clean up for debugging. Also work on event badge related.

Browse Source
This commit is contained in:
Scott Idem
2022-02-15 11:33:35 -05:00
parent 57b3b358ad
commit cfa58f2c32
6 changed files with 98 additions and 122 deletions
Download Patch File Download Diff File Expand all files Collapse all files

113
app/routers/event_badge.py
View File

@@ -3,9 +3,9 @@ from fastapi import APIRouter, Body, Depends, Header, HTTPException, Query, Resp
from pydantic import BaseModel, EmailStr, Field
from typing import Dict, List, Optional, Set, Union
from app.lib_general import log, logging
from app.lib_general import log, logging, common_route_params, Common_Route_Params
from app.config import settings
from app.db_sql import sql_insert, sql_update, sql_insert_or_update, sql_select, sql_delete, redis_lookup_id_random
from app.db_sql import sql_enable_part, sql_insert, sql_update, sql_insert_or_update, sql_limit_offset_part, sql_select, sql_delete, redis_lookup_id_random
from app.routers.api_crud import delete_obj_template, get_obj_template, get_obj_li_template, patch_obj_template, post_obj_template
@@ -135,70 +135,51 @@ async def get_event_badge_obj_li(
# ### BEGIN ### API Event Badge ### search_event_badge_obj_li() ###
# Updated 2021-07-28
# Updated 2022-02-15
@router.get('/event/{event_id}/badge/search', response_model=Resp_Body_Base)
async def search_event_badge_obj_li(
# account_id: str = Query(None, min_length=11, max_length=22),
event_id: str = Query(None, min_length=11, max_length=22),
event_badge_id: str = Query(None, min_length=3, max_length=22),
event_person_id: str = Query(None, min_length=3, max_length=22),
external_id: str = Query('%', max_length=50),
external_id: str = Query('%', max_length=100), # NOTE: Not currently used! 2022-02-15
given_name: str = Query('%', max_length=25),
family_name: str = Query('%', max_length=25),
email: str = Query('%', max_length=50),
limit: int = 50,
enabled: str = 'enabled',
x_account_id: str = Header(...),
by_alias: bool = True,
exclude_unset: bool = True,
response: Response = Response,
commons: Common_Route_Params = Depends(common_route_params),
):
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
if account_id := redis_lookup_id_random(record_id_random=x_account_id, table_name='account'): pass
else: return mk_resp(data=None, status_code=404, response=response)
# account_id = commons.x_account_id
if event_id := redis_lookup_id_random(record_id_random=event_id, table_name='event'): pass
else: return mk_resp(data=None, status_code=404, response=response)
else: return mk_resp(data=None, status_code=404, response=commons.response)
log.info(f'Searching: Event ID: {event_id}; Event Badge ID: {event_badge_id}; Event Person ID: {event_person_id}; External ID: {external_id}; Give Name: {given_name}; Family Name: {family_name}; Email: {email}')
data = {}
data['account_id'] = account_id
# data['account_id'] = account_id
data['event_id'] = event_id
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(event_badge_id)
if event_badge_id and len(event_badge_id) > 2:
log.info(f'Found event_badge_id: {event_badge_id}')
log.info(f'Using Event Badge ID: {event_badge_id}')
data['event_badge_id_random'] = f'%{event_badge_id}%'
elif event_person_id and len(event_person_id) > 2:
log.info(f'Found event_person_id: {event_person_id}')
log.info(f'Using Event Person ID: {event_person_id}')
data['event_person_id_random'] = f'%{event_person_id}%'
else:
log.info(f'Using other data fields')
data['external_id'] = '%'+external_id+'%'
data['given_name'] = '%'+given_name+'%'
data['family_name'] = '%'+family_name+'%'
data['email'] = '%'+email+'%' # Adding the % symbol here because it turns certain combinations into special characters
# data['from_datetime'] = from_datetime
# data['to_datetime'] = to_datetime
log.debug(data)
if enabled in ['enabled', 'disabled', 'all']:
if enabled == 'enabled':
data['enable'] = True
sql_enabled = f'AND `event_badge`.enable = :enable'
elif enabled == 'disabled':
data['enable'] = False
sql_enabled = f'AND `event_badge`.enable = :enable'
elif enabled == 'all':
sql_enabled = ''
if limit:
data['limit'] = limit
sql_limit = f'LIMIT :limit'
else:
sql_limit = ''
sql_enabled, data['enable'] = sql_enable_part(table_name='event_badge', enabled=commons.enabled) # Reasonably safe return str and bool
sql_limit = sql_limit_offset_part(limit=commons.limit, offset=commons.offset) # Reasonably safe return str
if event_badge_id and len(event_badge_id) > 2:
sql = f"""
@@ -244,51 +225,55 @@ async def search_event_badge_obj_li(
ORDER BY event_badge.given_name ASC, event_badge.family_name ASC, event_badge.created_on DESC, event_badge.updated_on DESC
{sql_limit};
"""
log.debug(sql)
if sql_result := sql_select(data=data, sql=sql, as_list=True, rm_id_random=False): # rm_id_random=True
if sql_result := sql_select(data=data, sql=sql, as_list=True, rm_id_random=False, log_lvl=logging.INFO): # rm_id_random=True
resp_data_li = []
for record in sql_result:
resp_data = Event_Badge_Base(**record).dict(by_alias=by_alias, exclude_unset=exclude_unset)
resp_data = Event_Badge_Base(**record).dict(by_alias=commons.by_alias, exclude_unset=commons.exclude_unset)
resp_data_li.append(resp_data)
return mk_resp(data=resp_data_li, response=response)
return mk_resp(data=resp_data_li, response=commons.response)
elif isinstance(sql_result, list) or sql_result is None: # Empty list or None
log.info('No results')
return mk_resp(data=None, status_code=404, response=commons.response) # Not Found
else:
log.debug(sql_result)
return mk_resp(data=None, status_code=404, response=response)
log.warning('Likely bad request')
return mk_resp(data=False, status_code=400, response=commons.response) # Bad Request
# ### END ### API Event Badge ### search_event_badge_obj_li() ###
# ### BEGIN ### API Event Badge ### get_event_badge_obj() ###
# Updated 2021-08-05
# Updated 2022-02-15
@router.get('/event/badge/{event_badge_id}', response_model=Resp_Body_Base)
async def get_event_badge_obj(
event_badge_id: str = Query(..., min_length=11, max_length=22),
limit: int = 5, # For now this covers any included objects or object lists
enabled: str = 'enabled', # For now this covers any included objects or object lists
inc_event_badge_template: bool = True,
x_account_id: str = Header(...),
by_alias: Optional[bool] = True,
exclude_unset: Optional[bool] = True,
response: Response = Response,
commons: Common_Route_Params = Depends(common_route_params),
):
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
if event_badge_id := redis_lookup_id_random(record_id_random=event_badge_id, table_name='event_badge'): pass
else: return mk_resp(data=None, status_code=404, response=response)
else: return mk_resp(data=None, status_code=404, response=commons.response)
event_badge_obj = load_event_badge_obj(
event_badge_id = event_badge_id,
limit = limit,
by_alias = by_alias,
exclude_unset = exclude_unset,
# model_as_dict = model_as_dict,
enabled = enabled,
inc_event_badge_template = inc_event_badge_template,
)
event_badge_dict = event_badge_obj.dict(by_alias=by_alias, exclude_unset=exclude_unset)
return mk_resp(data=event_badge_dict, response=response)
if event_badge_obj := load_event_badge_obj(
event_badge_id = event_badge_id,
inc_event_badge_template = inc_event_badge_template,
# limit = commons.limit,
# by_alias = commons.by_alias,
# exclude_unset = commons.exclude_unset,
# model_as_dict = model_as_dict,
# enabled = commons.enabled,
):
# event_badge_dict = event_badge_obj.dict(by_alias=commons.by_alias, exclude_unset=commons.exclude_unset)
# return mk_resp(data=event_badge_dict, response=commons.response)
response_data = event_badge_obj
else:
return mk_resp(data=False, status_code=400, response=commons.response) # Bad Request
return mk_resp(data=response_data, response=commons.response)
# ### END ### API Event Badge ### get_event_badge_obj() ###

6
app/routers/membership_person.py
View File

@@ -5,7 +5,7 @@ from typing import Dict, List, Optional, Set, Union
from app.lib_general import log, logging, common_route_params, Common_Route_Params
from app.config import settings
from app.db_sql import sql_insert, sql_update, sql_insert_or_update, sql_select, sql_delete, redis_lookup_id_random
from app.db_sql import sql_enable_part, sql_insert, sql_update, sql_insert_or_update, sql_limit_offset_part, sql_select, sql_delete, redis_lookup_id_random
from app.routers.api_crud import delete_obj_template, get_obj_template, get_obj_li_template, patch_obj_template, post_obj_template
@@ -401,7 +401,7 @@ async def lookup_membership_person_obj(
inc_membership_person_group_list = inc_membership_person_group_list,
inc_membership_person_profile = inc_membership_person_profile,
inc_membership_person_type = inc_membership_person_type,
).dict(by_alias=by_alias, exclude_unset=exclude_unset)
).dict(by_alias=commons.by_alias, exclude_unset=commons.exclude_unset)
data = membership_person_obj
elif isinstance(membership_person_obj_result, list):
membership_person_obj_li = []
@@ -414,7 +414,7 @@ async def lookup_membership_person_obj(
inc_membership_person_group_list = inc_membership_person_group_list,
inc_membership_person_profile = inc_membership_person_profile,
inc_membership_person_type = inc_membership_person_type,
).dict(by_alias=by_alias, exclude_unset=exclude_unset)
).dict(by_alias=commons.by_alias, exclude_unset=commons.exclude_unset)
)
data = membership_person_obj_li
else:

19
app/routers/person.py
View File

@@ -195,15 +195,15 @@ async def person_obj_external_id(
person_id = person_data.get('person_id')
if person_dict := load_person_obj(
person_id = person_id,
limit = commons.limit,
model_as_dict = True, # NOTE: returning model as a dict
enabled = commons.enabled,
inc_address = inc_address,
inc_contact = inc_contact,
inc_user = inc_user,
inc_user_role_list = inc_user_role_list,
):
person_id = person_id,
limit = commons.limit,
model_as_dict = True, # NOTE: returning model as a dict
enabled = commons.enabled,
inc_address = inc_address,
inc_contact = inc_contact,
inc_user = inc_user,
inc_user_role_list = inc_user_role_list,
):
if isinstance(person_dict, dict):
response_data = person_dict
else:
@@ -338,6 +338,7 @@ async def get_person_obj(
inc_post_list: bool = False, # Priority l2
inc_post_comment_list: bool = False, # Priority l3
inc_user: bool = False, # Priority l1
commons: Common_Route_Params = Depends(common_route_params),
):
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL