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

Work on Impexium, exhibit, and badge related

Browse Source
This commit is contained in:
Scott Idem
2022-02-23 18:34:21 -05:00
parent 5a84119c77
commit 18d8b73eb3
10 changed files with 489 additions and 346 deletions
Download Patch File Download Diff File Expand all files Collapse all files

189
app/methods/e_impexium_methods.py
View File

@@ -27,10 +27,11 @@ api['headers'] = { 'Content-Type': 'application/json;charset=UTF-8' }
# ### BEGIN ### API External Impexium Methods ### get_access_token() ### # ### BEGIN ### API External Impexium Methods ### get_access_token() ###
# Updated 2022-02-18 # Updated 2022-02-18
@logger_reset @logger_reset
def get_access_token(api, app): def get_access_token():
log.setLevel(logging.INFO) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals()) log.debug(locals())
log.debug(f'App data:\n{app}')
log.debug(f'API data:\n{api}') log.debug(f'API data:\n{api}')
endpoint = '/WebApiUrl' endpoint = '/WebApiUrl'
@@ -40,10 +41,10 @@ def get_access_token(api, app):
resp = requests.post(url=uri, json=data, headers=api['headers']) resp = requests.post(url=uri, json=data, headers=api['headers'])
log.debug('Status Code:', resp.status_code) log.debug(f'Status Code: {resp.status_code}')
log.debug('Headers:', resp.headers) log.debug(f'Headers: {resp.headers}')
# log.debug('Encoding:', resp.encoding) # log.debug(f'Encoding: {resp.encoding}')
log.debug('JSON:', resp.json()) log.debug(f'JSON: {resp.json()}')
# log.debug('Text:', resp.text) # log.debug('Text:', resp.text)
response_data = resp.json() response_data = resp.json()
@@ -62,29 +63,40 @@ def get_access_token(api, app):
# ### BEGIN ### API External Impexium Methods ### authenticate() ### # ### BEGIN ### API External Impexium Methods ### authenticate() ###
# Updated 2022-02-18 # Updated 2022-02-18
@logger_reset @logger_reset
def authenticate(api, app): def authenticate():
log.setLevel(logging.INFO) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals()) log.debug(locals())
result = get_access_token()
log.debug(result)
log.debug(f'App data:\n{app}')
log.debug(f'API data:\n{api}')
data = { 'AppId': app['id'], 'AppPassword': app['password'], 'AppUserEmail': app['user_email'], 'AppUserPassword': app['user_password'] } data = { 'AppId': app['id'], 'AppPassword': app['password'], 'AppUserEmail': app['user_email'], 'AppUserPassword': app['user_password'] }
resp = requests.post(url=api['auth_uri'], json=data, headers=api['headers']) try_request = True
limit = 0
while try_request and limit < 3:
limit = limit + 1
log.debug('Status Code:', resp.status_code) resp = requests.post(url=api['auth_uri'], json=data, headers=api['headers'])
log.debug('Headers:', resp.headers)
# log.debug('Encoding:', resp.encoding)
log.debug('JSON:', resp.json())
# log.debug('Text:', resp.text)
response_data = resp.json() log.debug(f'Status Code: {resp.status_code}')
log.debug(f'Headers: {resp.headers}')
# log.debug(f'Encoding: {resp.encoding}')
log.debug(f'JSON: {resp.json()}')
# log.debug('Text:', resp.text)
api['headers']['AppToken'] = response_data['appToken'] if resp:
api['headers']['UserToken'] = response_data['userToken'] try_request = False
api['base_url'] = response_data['uri'] response_data = resp.json()
api['headers']['AppToken'] = response_data['appToken']
# app['user_id'] = response_data['userId'] api['headers']['UserToken'] = response_data['userToken']
# app['sso_token'] = response_data['ssoToken'] api['base_url'] = response_data['uri']
# app['app_configuration'] = response_data['appConfiguration'] # allowGoogleLogin=False, allowLinkedInLogin=False else:
log.warning('Hit rate limit? Sleeping for .5 seconds...')
time.sleep(.5)
log.debug(api) log.debug(api)
@@ -92,11 +104,11 @@ def authenticate(api, app):
# ### END ### API External Impexium Methods ### authenticate() ### # ### END ### API External Impexium Methods ### authenticate() ###
api = get_access_token(api=api, app=app) # api = get_access_token(api=api, app=app)
log.debug(api) # log.debug(api)
api = authenticate(api=api, app=app) # api = authenticate(api=api, app=app)
log.debug(api) # log.debug(api)
# ### BEGIN ### API External Impexium Methods ### get_custom_fields() ### # ### BEGIN ### API External Impexium Methods ### get_custom_fields() ###
@@ -115,10 +127,10 @@ def get_custom_fields(api, name=None, page=1):
resp = requests.get(url=uri, params=params, headers=api['headers']) resp = requests.get(url=uri, params=params, headers=api['headers'])
log.debug('Status Code:', resp.status_code) log.debug(f'Status Code: {resp.status_code}')
log.debug('Headers:', resp.headers) log.debug(f'Headers: {resp.headers}')
# log.debug('Encoding:', resp.encoding) # log.debug(f'Encoding: {resp.encoding}')
log.debug('JSON:', resp.json()) log.debug(f'JSON: {resp.json()}')
# log.debug('Text:', resp.text) # log.debug('Text:', resp.text)
response_data = resp.json() response_data = resp.json()
@@ -145,10 +157,10 @@ def get_events(page=1):
resp = requests.get(url=uri, params=params, headers=api['headers']) resp = requests.get(url=uri, params=params, headers=api['headers'])
log.debug('Status Code:', resp.status_code) log.debug(f'Status Code: {resp.status_code}')
log.debug('Headers:', resp.headers) log.debug(f'Headers: {resp.headers}')
# log.debug('Encoding:', resp.encoding) # log.debug(f'Encoding: {resp.encoding}')
log.debug('JSON:', resp.json()) log.debug(f'JSON: {resp.json()}')
# log.debug('Text:', resp.text) # log.debug('Text:', resp.text)
response_data = resp.json() response_data = resp.json()
@@ -167,37 +179,64 @@ def get_events(page=1):
# Updated 2021-10-07 # Updated 2021-10-07
def get_event_registrants( def get_event_registrants(
event_code: str, event_code: str,
page: int = 1, page: int = 0,
return_all: bool = False # return_all: bool = False
): ):
log.setLevel(logging.INFO) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals()) log.debug(locals())
endpoint = f'/Events/{event_code}/Registrations/{page}' result = authenticate()
uri = api['base_url']+endpoint log.debug(result)
params = {} try_request = True
resp = requests.get(url=uri, params=params, headers=api['headers']) if page > 0:
loop_count = page
log.debug('Status Code:', resp.status_code) loop_limit = page + 1
log.debug('Headers:', resp.headers)
# log.debug('Encoding:', resp.encoding)
log.debug('JSON:', resp.json())
# log.debug('Text:', resp.text)
response_data = resp.json()
# pp = pprint.PrettyPrinter(indent=2)
# pp.pprint(response_data)
# print(response_data)
# print('**************************')
event_registrant_li = response_data
if return_all:
return event_registrant_li
else: else:
return event_registrant_li['dataList'] page = 1
loop_count = 0
loop_limit = 15
# current_page = 0
impexium_event_registration_list = []
while try_request and loop_count < loop_limit:
# if page == 0:
# page = 1
# # loop_limit = 10
# try_request = True
# else:
# # page = loop_count
# pass
endpoint = f'/Events/{event_code}/Registrations/{page}'
uri = api['base_url']+endpoint
params = {}
if resp := requests.get(url=uri, params=params, headers=api['headers']): pass
else: break
log.debug(f'Status Code: {resp.status_code}')
log.debug(f'Headers: {resp.headers}')
# log.debug(f'Encoding: {resp.encoding}')
# log.debug(f'JSON: {resp.json()}')
# log.debug('Text:', resp.text)
response_data = resp.json()
current_page = response_data.get('pageNumber')
log.info(f'Current Page: {current_page}')
if current_page >= 1:
impexium_event_registration_list = impexium_event_registration_list + response_data.get('dataList')
page = current_page + 1
log.warning('Avoiding rate limit. Sleeping for .1 seconds...')
time.sleep(.1)
else:
try_request = False
loop_count = loop_count + 1
return impexium_event_registration_list
# ### END ### API Impexium Methods ### get_event_registrants() ### # ### END ### API Impexium Methods ### get_event_registrants() ###
@@ -214,10 +253,10 @@ def get_individual_registrations(api, record_number, event_code=None, page=1):
resp = requests.get(url=uri, params=params, headers=api['headers']) resp = requests.get(url=uri, params=params, headers=api['headers'])
log.debug('Status Code:', resp.status_code) log.debug(f'Status Code: {resp.status_code}')
log.debug('Headers:', resp.headers) log.debug(f'Headers: {resp.headers}')
# log.debug('Encoding:', resp.encoding) # log.debug(f'Encoding: {resp.encoding}')
log.debug('JSON:', resp.json()) log.debug(f'JSON: {resp.json()}')
# log.debug('Text:', resp.text) # log.debug('Text:', resp.text)
response_data = resp.json() response_data = resp.json()
@@ -245,10 +284,10 @@ def get_individual_purchases(api, record_number, product_code=None, purchased_si
resp = requests.get(url=uri, params=params, headers=api['headers']) resp = requests.get(url=uri, params=params, headers=api['headers'])
log.debug('Status Code:', resp.status_code) log.debug(f'Status Code: {resp.status_code}')
log.debug('Headers:', resp.headers) log.debug(f'Headers: {resp.headers}')
# log.debug('Encoding:', resp.encoding) # log.debug(f'Encoding: {resp.encoding}')
log.debug('JSON:', resp.json()) log.debug(f'JSON: {resp.json()}')
# log.debug('Text:', resp.text) # log.debug('Text:', resp.text)
response_data = resp.json() response_data = resp.json()
@@ -276,10 +315,10 @@ def get_individual_custom_fields(api, individual_id):
resp = requests.get(url=uri, params=params, headers=api['headers']) resp = requests.get(url=uri, params=params, headers=api['headers'])
log.debug('Status Code:', resp.status_code) log.debug(f'Status Code: {resp.status_code}')
log.debug('Headers:', resp.headers) log.debug(f'Headers: {resp.headers}')
# log.debug('Encoding:', resp.encoding) # log.debug(f'Encoding: {resp.encoding}')
log.debug('JSON:', resp.json()) log.debug(f'JSON: {resp.json()}')
# log.debug('Text:', resp.text) # log.debug('Text:', resp.text)
response_data = resp.json() response_data = resp.json()
@@ -308,10 +347,10 @@ def get_individual_orders_open(api, record_number, include_line_items=False, fro
resp = requests.get(url=uri, params=params, headers=api['headers']) resp = requests.get(url=uri, params=params, headers=api['headers'])
log.debug('Status Code:', resp.status_code) log.debug(f'Status Code: {resp.status_code}')
log.debug('Headers:', resp.headers) log.debug(f'Headers: {resp.headers}')
# log.debug('Encoding:', resp.encoding) # log.debug(f'Encoding: {resp.encoding}')
log.debug('JSON:', resp.json()) log.debug(f'JSON: {resp.json()}')
# log.debug('Text:', resp.text) # log.debug('Text:', resp.text)
response_data = resp.json() response_data = resp.json()

86
app/methods/event_badge_methods.py
View File

@@ -91,3 +91,89 @@ def get_event_badge_template_id_w_event_id(
else: return False else: return False
else: return None else: return None
# ### END ### API Event Badge Methods ### get_event_badge_template_id_w_event_id() ### # ### END ### API Event Badge Methods ### get_event_badge_template_id_w_event_id() ###
# ### BEGIN ### API Event Badge Methods ### create_update_event_badge_obj_v4() ###
# Updated 2022-02-23
def create_update_event_badge_obj_v4(
event_badge_dict_obj: Event_Badge_Base|dict,
event_badge_id: int|str = None,
# account_id: int|str = None,
# event_id: int|str = None,
event_person_id: int|str = None,
create_sub_obj: bool = False,
fail_any: bool = False, # Fail if any thing goes wrong for sub objects
return_outline: bool = False,
) -> int|bool:
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
# ### SECTION ### Secondary data validation
log.info('Checking requirements...')
if event_badge_id:
log.info(f'Event Badge ID passed. Update existing Event Badge. Event Badge ID: {event_badge_id}')
if event_badge_id := redis_lookup_id_random(record_id_random=event_badge_id, table_name='event_badge'): pass
else:
log.error('Event Badge ID passed but is invalid. Failed requirement.')
return False
else:
log.info('No Event Badge ID passed. Create new Event Badge. Required: Event Person ID')
if event_person_id := redis_lookup_id_random(record_id_random=event_person_id, table_name='event_person'): pass
else:
log.error('Missing or invalid Event Person ID passed. Failed requirement.')
log.info(f'Event Person ID: {event_person_id}')
return False
log.info('Create dictionary or Pydantic object')
log.debug(type(event_badge_dict_obj))
if isinstance(event_badge_dict_obj, dict):
event_badge_dict = event_badge_dict_obj
if event_badge_id:
event_badge_dict['event_badge_id'] = event_badge_id
try:
event_badge_obj = Event_Badge_Base(**event_badge_dict)
except ValidationError as e:
log.error(e.json())
return False
else:
event_badge_obj = event_badge_dict_obj
if event_badge_id:
# NOTE: Can't update the ID alias if it was never set.
event_badge_obj.id = event_badge_id
log.debug(event_badge_obj)
event_badge_dict = event_badge_obj.dict(by_alias=False, exclude_defaults=False, exclude_unset=True, exclude={'event_badge_template', 'event_id', 'event_id_random', 'created_on', 'updated_on'})
# ### SECTION ### Process data
if event_person_id: pass
elif event_person_id := event_badge_obj.event_person_id: pass
if event_badge_id:
if event_badge_dict_up_result := sql_update(data=event_badge_dict, table_name='event_badge', rm_id_random=True): pass
else:
log.warning(f'Event Badge not updated. Event Badge ID: {event_badge_id}')
log.debug(event_badge_dict_up_result)
return False
log.debug(event_badge_dict_up_result)
else:
if event_badge_dict_in_result := sql_insert(data=event_badge_dict, table_name='event_badge', rm_id_random=True, id_random_length=8): pass
else:
log.warning(f'Event Badge not created.')
log.debug(event_badge_dict_in_result)
return False
log.debug(event_badge_dict_in_result)
event_badge_id = event_badge_dict_in_result
event_badge_outline = {}
event_badge_outline['event_badge_id'] = event_badge_id
if return_outline:
log.debug(f'Returning the Event Badge Outline: {event_badge_outline}')
return event_badge_outline
else:
log.debug(f'Returning the Event Badge ID: {event_badge_id}')
return event_badge_id
# ### END ### API Event Badge Methods ### create_update_event_badge_obj_v4() ###

115
app/methods/event_person_methods.py
View File

@@ -8,10 +8,10 @@ from app.db_sql import get_account_id_w_for_type_id, redis_lookup_id_random, sql
from app.lib_general import log, logging from app.lib_general import log, logging
# from app.methods.event_abstract_methods import load_event_abstract_obj # from app.methods.event_abstract_methods import load_event_abstract_obj
from app.methods.event_badge_methods import load_event_badge_obj from app.methods.event_badge_methods import create_update_event_badge_obj_v4, load_event_badge_obj
# from app.methods.event_exhibit_methods import load_event_exhibit_obj # from app.methods.event_exhibit_methods import load_event_exhibit_obj
# from app.methods.event_file_methods import load_event_file_obj # from app.methods.event_file_methods import load_event_file_obj
from app.methods.event_person_profile_methods import load_event_person_profile_obj from app.methods.event_person_profile_methods import create_update_event_person_profile_obj_v4, load_event_person_profile_obj
# from app.methods.event_presentation_methods import load_event_presentation_obj # from app.methods.event_presentation_methods import load_event_presentation_obj
# from app.methods.event_presenter_methods import get_event_presenter_rec_list, load_event_presenter_obj # from app.methods.event_presenter_methods import get_event_presenter_rec_list, load_event_presenter_obj
# from app.methods.event_registration_methods import create_event_registration_obj, load_event_registration_obj, update_event_registration_obj_v3 # from app.methods.event_registration_methods import create_event_registration_obj, load_event_registration_obj, update_event_registration_obj_v3
@@ -308,16 +308,15 @@ def get_event_id_w_event_person_id(
# ### BEGIN ### API Event Person Methods ### create_update_event_person_obj_v4() ### # ### BEGIN ### API Event Person Methods ### create_update_event_person_obj_v4() ###
# NOTE: This will create or update a event_person. # Updated 2022-02-23
# Rewrite and updated 2021-08-25
def create_update_event_person_obj_v4( def create_update_event_person_obj_v4(
event_person_dict_obj: Event_Person_Base|dict, event_person_dict_obj: Event_Person_Base|dict,
event_person_id: int|str|None = None, event_person_id: int|str|None = None,
account_id: int|str|None = None, account_id: int|str|None = None,
event_id: int|str|None = None, event_id: int|str|None = None,
event_badge_id: int|None = None,
event_person_profile_id: int|None = None,
create_sub_obj: bool = False, create_sub_obj: bool = False,
fail_any: bool = False, # Fail if any thing goes wrong for sub objects fail_any: bool = False, # Fail if any thing goes wrong for sub objects
return_outline: bool = False, return_outline: bool = False,
@@ -325,6 +324,7 @@ def create_update_event_person_obj_v4(
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals()) log.debug(locals())
# ### SECTION ### Secondary data validation
log.info('Checking requirements...') log.info('Checking requirements...')
if event_person_id: if event_person_id:
log.info(f'Event Person ID passed. Update existing Event Person. Event Person ID: {event_person_id}') log.info(f'Event Person ID passed. Update existing Event Person. Event Person ID: {event_person_id}')
@@ -377,6 +377,7 @@ def create_update_event_person_obj_v4(
log.error('Unable to get Account ID from related object.') log.error('Unable to get Account ID from related object.')
return False return False
log.info('Create dictionary or Pydantic object')
log.debug(type(event_person_dict_obj)) log.debug(type(event_person_dict_obj))
if isinstance(event_person_dict_obj, dict): if isinstance(event_person_dict_obj, dict):
event_person_dict = event_person_dict_obj event_person_dict = event_person_dict_obj
@@ -388,8 +389,6 @@ def create_update_event_person_obj_v4(
event_person_dict['event_id'] = event_id event_person_dict['event_id'] = event_id
try: try:
event_person_obj = Event_Person_Base(**event_person_dict) event_person_obj = Event_Person_Base(**event_person_dict)
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(event_person_obj)
except ValidationError as e: except ValidationError as e:
log.error(e.json()) log.error(e.json())
return False return False
@@ -402,8 +401,22 @@ def create_update_event_person_obj_v4(
event_person_obj.account_id = account_id event_person_obj.account_id = account_id
if event_id: if event_id:
event_person_obj.event_id = event_id event_person_obj.event_id = event_id
log.debug(event_person_obj)
event_person_dict = event_person_obj.dict(by_alias=False, exclude_defaults=False, exclude_unset=True, exclude={'event_badge', 'event_registration', 'created_on', 'updated_on'}) event_person_dict = event_person_obj.dict(by_alias=False, exclude_defaults=False, exclude_unset=True, exclude={'event_badge', 'event_person_profile', 'event_registration', 'created_on', 'updated_on'})
# ### SECTION ### Process data
if not event_badge_id:
if event_person_obj.event_badge and event_person_obj.event_badge.id:
event_badge_id = event_person_obj.event_badge.id
elif event_person_obj.event_badge and event_person_obj.event_badge_id:
event_badge_id = event_person_obj.event_badge_id
if not event_person_profile_id:
if event_person_obj.event_person_profile and event_person_obj.event_person_profile.id:
event_person_profile_id = event_person_obj.event_person_profile.id
elif event_person_obj.event_person_profile and event_person_obj.event_person_profile_id:
event_person_profile_id = event_person_obj.event_person_profile_id
if event_person_id: if event_person_id:
if event_person_dict_up_result := sql_update(data=event_person_dict, table_name='event_person', rm_id_random=True): pass if event_person_dict_up_result := sql_update(data=event_person_dict, table_name='event_person', rm_id_random=True): pass
@@ -425,49 +438,54 @@ def create_update_event_person_obj_v4(
event_person_outline = {} event_person_outline = {}
event_person_outline['event_person_id'] = event_person_id event_person_outline['event_person_id'] = event_person_id
# NOTE: WARNING: This has not been fully tested yet. 2021-08-25 # Updated 2022-02-23
# NOTE: Use object model version because of better type checking and validations if event_person_obj.event_badge:
# if event_person_obj.event_badge: if event_badge_id: new_record = False
# event_person_outline['event_badge_id'] = None else: new_record = True
# event_badge_obj = event_person_obj.event_badge event_person_outline['event_badge_id'] = None
# if event_badge_id := event_person_obj.event_badge_id: pass event_badge_obj = event_person_obj.event_badge
# elif event_badge_id := event_badge_obj.id: pass event_badge_obj.event_person_id = event_person_id
# else: event_badge_id = None create_update_event_badge_obj_result = create_update_event_badge_obj_v4(
# # event_badge_obj.id event_badge_dict_obj = event_badge_obj,
# event_badge_obj.event_person_id = event_person_id event_badge_id = event_badge_id,
# create_update_event_badge_obj_result = create_update_event_badge_obj_v4( # account_id = account_id,
# event_badge_dict_obj = event_badge_obj, # event_id = event_id,
# event_badge_id = event_badge_id, event_person_id = event_person_id,
# # account_id = account_id, fail_any = fail_any,
# event_id = event_id, return_outline = return_outline,
# event_person_id = event_person_id, )
# fail_any = fail_any, if isinstance(create_update_event_badge_obj_result, int):
# return_outline = return_outline, event_badge_id = create_update_event_badge_obj_result
# ) elif create_update_event_badge_obj_result == True: pass
# if isinstance(create_update_event_badge_obj_result, int): else:
# event_badge_id = create_update_event_badge_obj_result log.warning(f'Create or Update failed while trying create_update_event_badge_obj_v4(): {create_update_event_badge_obj_result}')
# elif create_update_event_badge_obj_result == True: pass event_badge_id = None
# else:
# log.warning(f'Create or Update failed while trying create_update_event_badge_obj_v4(): {create_update_event_badge_obj_result}')
# event_badge_id = None
# event_person_outline['event_badge_id'] = event_badge_id # NOTE: Because this is a new record and this is a 1 to 1 link, we need to update the event_person record with the new event_person_profile_id.
if new_record:
event_person_data = {}
event_person_data['event_badge_id'] = event_badge_id
if event_person_data_up_result := sql_update(data=event_person_data, table_name='event_person', record_id=event_person_id): pass
else:
log.warning(f'Event Person not updated with new Event Badge ID. Event Person ID: {event_person_id}; Event Badge ID: {event_badge_id}')
log.debug(event_person_data_up_result)
return False
log.debug(event_person_data_up_result)
# NOTE: WARNING: This has not been fully tested yet. 2021-09-07 event_person_outline['event_badge_id'] = event_badge_id
# NOTE: Use object model version because of better type checking and validations
# Updated 2022-02-23
if event_person_obj.event_person_profile: if event_person_obj.event_person_profile:
if event_person_profile_id: new_record = False
else: new_record = True
event_person_outline['event_person_profile_id'] = None event_person_outline['event_person_profile_id'] = None
event_person_profile_obj = event_person_obj.event_person_profile event_person_profile_obj = event_person_obj.event_person_profile
if event_person_profile_id := event_person_obj.event_person_profile_id: pass
elif event_person_profile_id := event_person_profile_obj.id: pass
else: event_person_profile_id = None
# event_person_profile_obj.id
event_person_profile_obj.event_person_id = event_person_id event_person_profile_obj.event_person_id = event_person_id
create_update_event_person_profile_obj_result = create_update_event_person_profile_obj_v4( create_update_event_person_profile_obj_result = create_update_event_person_profile_obj_v4(
event_person_profile_dict_obj = event_person_profile_obj, event_person_profile_dict_obj = event_person_profile_obj,
event_person_profile_id = event_person_profile_id, event_person_profile_id = event_person_profile_id,
account_id = account_id, # account_id = account_id,
event_id = event_id, # event_id = event_id,
event_person_id = event_person_id, event_person_id = event_person_id,
fail_any = fail_any, fail_any = fail_any,
return_outline = return_outline, return_outline = return_outline,
@@ -479,6 +497,17 @@ def create_update_event_person_obj_v4(
log.warning(f'Create or Update failed while trying create_update_event_person_profile_obj_v4(): {create_update_event_person_profile_obj_result}') log.warning(f'Create or Update failed while trying create_update_event_person_profile_obj_v4(): {create_update_event_person_profile_obj_result}')
event_person_profile_id = None event_person_profile_id = None
# NOTE: Because this is a new record and this is a 1 to 1 link, we need to update the event_person record with the new event_person_profile_id.
if new_record:
event_person_data = {}
event_person_data['event_person_profile_id'] = event_person_profile_id
if event_person_data_up_result := sql_update(data=event_person_data, table_name='event_person', record_id=event_person_id): pass
else:
log.warning(f'Event Person not updated with new Event Person Profile ID. Event Person ID: {event_person_id}; Event Person Profile ID: {event_person_profile_id}')
log.debug(event_person_data_up_result)
return False
log.debug(event_person_data_up_result)
event_person_outline['event_person_profile_id'] = event_person_profile_id event_person_outline['event_person_profile_id'] = event_person_profile_id
if return_outline: if return_outline:

43
app/methods/event_person_profile_methods.py
View File

@@ -94,13 +94,12 @@ def get_event_person_id_w_event_person_profile_id(
# ### BEGIN ### API Event Person Profile Methods ### create_update_event_person_profile_obj_v4() ### # ### BEGIN ### API Event Person Profile Methods ### create_update_event_person_profile_obj_v4() ###
# NOTE: This will create or update a event_person_profile. # Updated 2022-02-23
# Rewrite and updated 2021-09-07
def create_update_event_person_profile_obj_v4( def create_update_event_person_profile_obj_v4(
event_person_profile_dict_obj: Event_Person_Profile_Base|dict, event_person_profile_dict_obj: Event_Person_Profile_Base|dict,
event_person_profile_id: int|str = None, event_person_profile_id: int|str = None,
account_id: int|str = None, # account_id: int|str = None,
event_id: int|str = None, # event_id: int|str = None,
event_person_id: int|str = None, event_person_id: int|str = None,
create_sub_obj: bool = False, create_sub_obj: bool = False,
fail_any: bool = False, # Fail if any thing goes wrong for sub objects fail_any: bool = False, # Fail if any thing goes wrong for sub objects
@@ -109,12 +108,8 @@ def create_update_event_person_profile_obj_v4(
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals()) log.debug(locals())
# ### SECTION ### Secondary data validation
log.info('Checking requirements...') log.info('Checking requirements...')
if account_id := redis_lookup_id_random(record_id_random=account_id, table_name='account'): pass
else:
log.error('Missing or invalid Account ID passed. Not required. Ignoring for now.')
log.info(f'Account ID: {account_id}')
if event_person_profile_id: if event_person_profile_id:
log.info(f'Event Person Profile ID passed. Update existing Event Person Profile. Event Person Profile ID: {event_person_profile_id}') log.info(f'Event Person Profile ID passed. Update existing Event Person Profile. Event Person Profile ID: {event_person_profile_id}')
@@ -131,6 +126,7 @@ def create_update_event_person_profile_obj_v4(
log.info(f'Event Person ID: {event_person_id}') log.info(f'Event Person ID: {event_person_id}')
return False return False
log.info('Create dictionary or Pydantic object')
log.debug(type(event_person_profile_dict_obj)) log.debug(type(event_person_profile_dict_obj))
if isinstance(event_person_profile_dict_obj, dict): if isinstance(event_person_profile_dict_obj, dict):
event_person_profile_dict = event_person_profile_dict_obj event_person_profile_dict = event_person_profile_dict_obj
@@ -138,8 +134,6 @@ def create_update_event_person_profile_obj_v4(
event_person_profile_dict['event_person_profile_id'] = event_person_profile_id event_person_profile_dict['event_person_profile_id'] = event_person_profile_id
try: try:
event_person_profile_obj = Event_Person_Profile_Base(**event_person_profile_dict) event_person_profile_obj = Event_Person_Profile_Base(**event_person_profile_dict)
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(event_person_profile_obj)
except ValidationError as e: except ValidationError as e:
log.error(e.json()) log.error(e.json())
return False return False
@@ -148,8 +142,16 @@ def create_update_event_person_profile_obj_v4(
if event_person_profile_id: if event_person_profile_id:
# NOTE: Can't update the ID alias if it was never set. # NOTE: Can't update the ID alias if it was never set.
event_person_profile_obj.id = event_person_profile_id event_person_profile_obj.id = event_person_profile_id
log.debug(event_person_profile_obj)
event_person_profile_dict = event_person_profile_obj.dict(by_alias=False, exclude_defaults=False, exclude_unset=True, exclude={'account_id', 'account_id_random', 'contact', 'event_cfg', 'event_id', 'event_id_random', 'organization', 'created_on', 'updated_on'}) event_person_profile_dict = event_person_profile_obj.dict(by_alias=False, exclude_defaults=False, exclude_unset=True, exclude={'account_id', 'account_id_random', 'contact', 'event_cfg', 'event_id', 'event_id_random', 'event_person_id', 'event_person_id_random', 'organization', 'created_on', 'updated_on'})
# ### SECTION ### Process data
contact_id = None
if event_person_profile_obj.contact and event_person_profile_obj.contact.id:
contact_id = event_person_profile_obj.contact.id
elif event_person_profile_obj.contact and event_person_profile_obj.contact_id:
contact_id = event_person_profile_obj.contact_id
if event_person_profile_id: if event_person_profile_id:
if event_person_profile_dict_up_result := sql_update(data=event_person_profile_dict, table_name='event_person_profile', rm_id_random=True): pass if event_person_profile_dict_up_result := sql_update(data=event_person_profile_dict, table_name='event_person_profile', rm_id_random=True): pass
@@ -169,13 +171,14 @@ def create_update_event_person_profile_obj_v4(
event_person_profile_id = event_person_profile_dict_in_result event_person_profile_id = event_person_profile_dict_in_result
# NOTE: This is really only needed if a new contact is being created # NOTE: This is really only needed if a new contact is being created
if not account_id:
log.info('Attempting to get Account ID from related object.') log.info('Attempting to get Account ID from related object.')
if account_id := event_person_profile_obj.account_id: pass if account_id := event_person_profile_obj.account_id: pass
elif account_id := event_person_profile_obj.account_id_random: pass # elif account_id := event_person_profile_obj.account_id_random: pass
elif event_person_id: pass else:
if event_person_id: pass
elif event_person_id := event_person_profile_obj.event_person_id: pass elif event_person_id := event_person_profile_obj.event_person_id: pass
elif event_person_id := event_person_profile_obj.event_person_id_random: pass # elif event_person_id := event_person_profile_obj.event_person_id_random: pass
if event_person_id: if event_person_id:
if account_id := get_account_id_w_for_type_id(for_type='event_person', for_id=event_person_id): pass if account_id := get_account_id_w_for_type_id(for_type='event_person', for_id=event_person_id): pass
@@ -190,10 +193,6 @@ def create_update_event_person_profile_obj_v4(
if event_person_profile_obj.contact: if event_person_profile_obj.contact:
event_person_profile_outline['contact_id'] = None event_person_profile_outline['contact_id'] = None
contact_obj = event_person_profile_obj.contact contact_obj = event_person_profile_obj.contact
if contact_id := event_person_profile_obj.contact_id: pass
elif contact_id := contact_obj.id: pass
else: contact_id = None
contact_obj.id
contact_obj.for_type = 'event_person_profile' contact_obj.for_type = 'event_person_profile'
contact_obj.for_id = event_person_profile_id contact_obj.for_id = event_person_profile_id
create_update_contact_obj_result = create_update_contact_obj_v4( create_update_contact_obj_result = create_update_contact_obj_v4(

38
app/models/event_badge_models.py
View File

@@ -1,4 +1,3 @@
from __future__ import annotations
import datetime, hashlib, logging, os, pytz, redis, secrets import datetime, hashlib, logging, os, pytz, redis, secrets
from typing import Dict, List, Optional, Set, Union from typing import Dict, List, Optional, Set, Union
@@ -37,6 +36,8 @@ class Event_Badge_Base(BaseModel):
person_id_random: Optional[str] person_id_random: Optional[str]
person_id: Optional[int] person_id: Optional[int]
external_id: Optional[str]
pronouns: Optional[str] # Preferred pronouns pronouns: Optional[str] # Preferred pronouns
informal_name: Optional[str] informal_name: Optional[str]
@@ -119,44 +120,33 @@ class Event_Badge_Base(BaseModel):
@validator('id', always=True) @validator('id', always=True)
def event_badge_id_lookup(cls, v, values, **kwargs): def event_badge_id_lookup(cls, v, values, **kwargs):
log.setLevel(logging.WARNING) if isinstance(v, int) and v > 0: return v
log.debug(locals()) elif id_random := values.get('id_random'):
return redis_lookup_id_random(record_id_random=id_random, table_name='event_badge')
if values['id_random']:
log.debug(values['id_random'])
return redis_lookup_id_random(record_id_random=values['id_random'], table_name='event_badge')
return None return None
@validator('event_id', always=True) @validator('event_id', always=True)
def event_id_lookup(cls, v, values, **kwargs): def event_id_lookup(cls, v, values, **kwargs):
log.setLevel(logging.WARNING) if isinstance(v, int) and v > 0: return v
log.debug(locals()) elif id_random := values.get('event_id_random'):
return redis_lookup_id_random(record_id_random=id_random, table_name='event')
if values['event_id_random']:
return redis_lookup_id_random(record_id_random=values['event_id_random'], table_name='event')
return None return None
@validator('event_person_id', always=True) @validator('event_person_id', always=True)
def event_person_id_lookup(cls, v, values, **kwargs): def event_person_id_lookup(cls, v, values, **kwargs):
log.setLevel(logging.WARNING) if isinstance(v, int) and v > 0: return v
log.debug(locals()) elif id_random := values.get('event_person_id_random'):
return redis_lookup_id_random(record_id_random=id_random, table_name='event_person')
if values['event_person_id_random']:
return redis_lookup_id_random(record_id_random=values['event_person_id_random'], table_name='event_person')
return None return None
@validator('person_id', always=True) @validator('person_id', always=True)
def person_id_lookup(cls, v, values, **kwargs): def person_id_lookup(cls, v, values, **kwargs):
log.setLevel(logging.WARNING) if isinstance(v, int) and v > 0: return v
log.debug(locals()) elif id_random := values.get('person_id_random'):
return redis_lookup_id_random(record_id_random=id_random, table_name='person')
if values['person_id_random']:
return redis_lookup_id_random(record_id_random=values['person_id_random'], table_name='person')
return None return None
class Config: class Config:
underscore_attrs_are_private = True underscore_attrs_are_private = True
allow_population_by_field_name = True allow_population_by_field_name = True
fields = base_fields fields = base_fields
Event_Badge_Base.update_forward_refs()

18
app/models/event_badge_template_models.py
View File

@@ -1,4 +1,3 @@
from __future__ import annotations
import datetime, hashlib, logging, os, pytz, redis, secrets import datetime, hashlib, logging, os, pytz, redis, secrets
from typing import Dict, List, Optional, Set, Union from typing import Dict, List, Optional, Set, Union
@@ -81,21 +80,16 @@ class Event_Badge_Template_Base(BaseModel):
@validator('id', always=True) @validator('id', always=True)
def event_badge_template_id_lookup(cls, v, values, **kwargs): def event_badge_template_id_lookup(cls, v, values, **kwargs):
log.setLevel(logging.WARNING) if isinstance(v, int) and v > 0: return v
log.debug(locals()) elif id_random := values.get('id_random'):
return redis_lookup_id_random(record_id_random=id_random, table_name='event_badge_template')
if values['id_random']:
log.debug(values['id_random'])
return redis_lookup_id_random(record_id_random=values['id_random'], table_name='event_badge_template')
return None return None
@validator('event_id', always=True) @validator('event_id', always=True)
def event_id_lookup(cls, v, values, **kwargs): def event_id_lookup(cls, v, values, **kwargs):
log.setLevel(logging.WARNING) if isinstance(v, int) and v > 0: return v
log.debug(locals()) elif id_random := values.get('event_id_random'):
return redis_lookup_id_random(record_id_random=id_random, table_name='event')
if values['event_id_random']:
return redis_lookup_id_random(record_id_random=values['event_id_random'], table_name='event')
return None return None
class Config: class Config:

124
app/models/event_person_models.py
View File

@@ -1,4 +1,3 @@
from __future__ import annotations
import datetime, hashlib, logging, os, pytz, redis, secrets import datetime, hashlib, logging, os, pytz, redis, secrets
from typing import Dict, List, Optional, Set, Union from typing import Dict, List, Optional, Set, Union
@@ -56,6 +55,8 @@ class Event_Person_Base(BaseModel):
user_id_random: Optional[str] user_id_random: Optional[str]
user_id: Optional[int] user_id: Optional[int]
external_id: Optional[str]
file_count: Optional[int] file_count: Optional[int]
priority: Optional[bool] priority: Optional[bool]
@@ -121,102 +122,78 @@ class Event_Person_Base(BaseModel):
#@validator('event_person_id_random', always=True) #@validator('event_person_id_random', always=True)
def event_person_id_random_copy(cls, v, values, **kwargs): def event_person_id_random_copy(cls, v, values, **kwargs):
log.setLevel(logging.WARNING)
log.debug(locals())
if values['id_random']: if values['id_random']:
return values['id_random'] return values['id_random']
return None return None
@validator('id', always=True) @validator('id', always=True)
def event_person_id_lookup(cls, v, values, **kwargs): def event_person_id_lookup(cls, v, values, **kwargs):
log.setLevel(logging.WARNING) if isinstance(v, int) and v > 0: return v
log.debug(locals()) elif id_random := values.get('id_random'):
return redis_lookup_id_random(record_id_random=id_random, table_name='event_person')
if values['id_random']:
log.debug(values['id_random'])
return redis_lookup_id_random(record_id_random=values['id_random'], table_name='event_person')
return None return None
@validator('account_id', always=True) @validator('account_id', always=True)
def account_id_lookup(cls, v, values, **kwargs): def account_id_lookup(cls, v, values, **kwargs):
log.setLevel(logging.WARNING) if isinstance(v, int) and v > 0: return v
log.debug(locals()) elif id_random := values.get('account_id_random'):
return redis_lookup_id_random(record_id_random=id_random, table_name='account')
if values['account_id_random']:
return redis_lookup_id_random(record_id_random=values['account_id_random'], table_name='account')
return None return None
@validator('event_id', always=True) @validator('event_id', always=True)
def event_id_lookup(cls, v, values, **kwargs): def event_id_lookup(cls, v, values, **kwargs):
log.setLevel(logging.WARNING) if isinstance(v, int) and v > 0: return v
log.debug(locals()) elif id_random := values.get('event_id_random'):
return redis_lookup_id_random(record_id_random=id_random, table_name='event')
if values['event_id_random']:
return redis_lookup_id_random(record_id_random=values['event_id_random'], table_name='event')
return None return None
@validator('event_badge_id', always=True) @validator('event_badge_id', always=True)
def event_badge_id_lookup(cls, v, values, **kwargs): def event_badge_id_lookup(cls, v, values, **kwargs):
log.setLevel(logging.WARNING) if isinstance(v, int) and v > 0: return v
log.debug(locals()) elif id_random := values.get('event_badge_id_random'):
return redis_lookup_id_random(record_id_random=id_random, table_name='event_badge')
if values['event_badge_id_random']:
return redis_lookup_id_random(record_id_random=values['event_badge_id_random'], table_name='event_badge')
return None return None
@validator('event_badge_vendor_id', always=True) @validator('event_badge_vendor_id', always=True)
def event_badge_vendor_id_lookup(cls, v, values, **kwargs): def event_badge_vendor_id_lookup(cls, v, values, **kwargs):
log.setLevel(logging.WARNING) if isinstance(v, int) and v > 0: return v
log.debug(locals()) elif id_random := values.get('event_badge_vendor_id_random'):
return redis_lookup_id_random(record_id_random=id_random, table_name='event_badge')
if values['event_badge_vendor_id_random']:
return redis_lookup_id_random(record_id_random=values['event_badge_vendor_id_random'], table_name='event_badge')
return None return None
@validator('event_badge_vip_id', always=True) @validator('event_badge_vip_id', always=True)
def event_badge_vip_id_lookup(cls, v, values, **kwargs): def event_badge_vip_id_lookup(cls, v, values, **kwargs):
log.setLevel(logging.WARNING) if isinstance(v, int) and v > 0: return v
log.debug(locals()) elif id_random := values.get('event_badge_vip_id_random'):
return redis_lookup_id_random(record_id_random=id_random, table_name='event_badge')
if values['event_badge_vip_id_random']:
return redis_lookup_id_random(record_id_random=values['event_badge_vip_id_random'], table_name='event_badge')
return None return None
@validator('event_person_profile_id', always=True) @validator('event_person_profile_id', always=True)
def event_person_profile_id_lookup(cls, v, values, **kwargs): def event_person_profile_id_lookup(cls, v, values, **kwargs):
log.setLevel(logging.WARNING) if isinstance(v, int) and v > 0: return v
log.debug(locals()) elif id_random := values.get('event_person_profile_id_random'):
return redis_lookup_id_random(record_id_random=id_random, table_name='event_person_profile')
if values['event_person_profile_id_random']:
return redis_lookup_id_random(record_id_random=values['event_person_profile_id_random'], table_name='event_person_profile')
return None return None
@validator('event_registration_id', always=True) @validator('event_registration_id', always=True)
def event_registration_id_lookup(cls, v, values, **kwargs): def event_registration_id_lookup(cls, v, values, **kwargs):
log.setLevel(logging.WARNING) if isinstance(v, int) and v > 0: return v
log.debug(locals()) elif id_random := values.get('event_registration_id_random'):
return redis_lookup_id_random(record_id_random=id_random, table_name='event_registration')
if values['event_registration_id_random']:
return redis_lookup_id_random(record_id_random=values['event_registration_id_random'], table_name='event_registration')
return None return None
@validator('person_id', always=True) @validator('person_id', always=True)
def person_id_lookup(cls, v, values, **kwargs): def person_id_lookup(cls, v, values, **kwargs):
log.setLevel(logging.WARNING) if isinstance(v, int) and v > 0: return v
log.debug(locals()) elif id_random := values.get('person_id_random'):
return redis_lookup_id_random(record_id_random=id_random, table_name='person')
if values['person_id_random']:
return redis_lookup_id_random(record_id_random=values['person_id_random'], table_name='person')
return None return None
@validator('user_id', always=True) @validator('user_id', always=True)
def user_id_lookup(cls, v, values, **kwargs): def user_id_lookup(cls, v, values, **kwargs):
log.setLevel(logging.WARNING) if isinstance(v, int) and v > 0: return v
log.debug(locals()) elif id_random := values.get('user_id_random'):
return redis_lookup_id_random(record_id_random=id_random, table_name='user')
if values['user_id_random']:
return redis_lookup_id_random(record_id_random=values['user_id_random'], table_name='user')
return None return None
class Config: class Config:
@@ -263,50 +240,31 @@ class Event_Person_New_Base(BaseModel):
#@validator('event_person_id_random', always=True) #@validator('event_person_id_random', always=True)
def event_person_id_random_copy(cls, v, values, **kwargs): def event_person_id_random_copy(cls, v, values, **kwargs):
log.setLevel(logging.WARNING)
log.debug(locals())
if values['id_random']: if values['id_random']:
return values['id_random'] return values['id_random']
return None return None
@validator('id', always=True) @validator('id', always=True)
def event_person_id_lookup(cls, v, values, **kwargs): def event_person_id_lookup(cls, v, values, **kwargs):
log.setLevel(logging.WARNING) if isinstance(v, int) and v > 0: return v
log.debug(locals()) elif id_random := values.get('id_random'):
return redis_lookup_id_random(record_id_random=id_random, table_name='event_person')
if values['id_random']:
log.debug(values['id_random'])
return redis_lookup_id_random(record_id_random=values['id_random'], table_name='event_person')
return None return None
@validator('account_id', always=True) @validator('account_id', always=True)
def account_id_lookup(cls, v, values, **kwargs): def account_id_lookup(cls, v, values, **kwargs):
log.setLevel(logging.WARNING) if isinstance(v, int) and v > 0: return v
log.debug(locals()) elif id_random := values.get('account_id_random'):
return redis_lookup_id_random(record_id_random=id_random, table_name='account')
if values['account_id_random']:
return redis_lookup_id_random(record_id_random=values['account_id_random'], table_name='account')
return None return None
@validator('event_id', always=True) @validator('event_id', always=True)
def event_id_lookup(cls, v, values, **kwargs): def event_id_lookup(cls, v, values, **kwargs):
log.setLevel(logging.WARNING) if isinstance(v, int) and v > 0: return v
log.debug(locals()) elif id_random := values.get('event_id_random'):
return redis_lookup_id_random(record_id_random=id_random, table_name='event')
if values['event_id_random']:
return redis_lookup_id_random(record_id_random=values['event_id_random'], table_name='event')
return None return None
# @validator('password', always=True)
# def hash_new_password(cls, v, values, **kwargs):
# log.setLevel(logging.WARNING)
# log.debug(locals())
# if values['new_password']:
# return secure_hash_string(string=values['new_password'])
# return None
class Config: class Config:
underscore_attrs_are_private = True underscore_attrs_are_private = True
allow_population_by_field_name = True allow_population_by_field_name = True

39
app/models/event_person_profile_models.py
View File

@@ -1,4 +1,3 @@
from __future__ import annotations
import datetime, hashlib, logging, os, pytz, redis, secrets import datetime, hashlib, logging, os, pytz, redis, secrets
from typing import Dict, List, Optional, Set, Union from typing import Dict, List, Optional, Set, Union
@@ -14,7 +13,7 @@ from app.models.organization_models import Organization_Base
class Event_Person_Profile_Base(BaseModel): class Event_Person_Profile_Base(BaseModel):
log.setLevel(logging.INFO) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals()) log.debug(locals())
id_random: Optional[str] = Field( id_random: Optional[str] = Field(
@@ -112,48 +111,36 @@ class Event_Person_Profile_Base(BaseModel):
#@validator('event_person_profile_id_random', always=True) #@validator('event_person_profile_id_random', always=True)
def event_person_profile_id_random_copy(cls, v, values, **kwargs): def event_person_profile_id_random_copy(cls, v, values, **kwargs):
log.setLevel(logging.WARNING)
log.debug(locals())
if values['id_random']: if values['id_random']:
return values['id_random'] return values['id_random']
return None return None
@validator('id', always=True) @validator('id', always=True)
def event_person_profile_id_lookup(cls, v, values, **kwargs): def event_person_profile_id_lookup(cls, v, values, **kwargs):
log.setLevel(logging.WARNING) if isinstance(v, int) and v > 0: return v
log.debug(locals()) elif id_random := values.get('id_random'):
return redis_lookup_id_random(record_id_random=id_random, table_name='event_person_profile')
if values['id_random']:
log.debug(values['id_random'])
return redis_lookup_id_random(record_id_random=values['id_random'], table_name='event_person_profile')
return None return None
@validator('account_id', always=True) @validator('account_id', always=True)
def account_id_lookup(cls, v, values, **kwargs): def account_id_lookup(cls, v, values, **kwargs):
log.setLevel(logging.WARNING) if isinstance(v, int) and v > 0: return v
log.debug(locals()) elif id_random := values.get('account_id_random'):
return redis_lookup_id_random(record_id_random=id_random, table_name='account')
if values['account_id_random']:
return redis_lookup_id_random(record_id_random=values['account_id_random'], table_name='account')
return None return None
@validator('contact_id', always=True) @validator('contact_id', always=True)
def contact_id_lookup(cls, v, values, **kwargs): def contact_id_lookup(cls, v, values, **kwargs):
log.setLevel(logging.WARNING) if isinstance(v, int) and v > 0: return v
log.debug(locals()) elif id_random := values.get('contact_id_random'):
return redis_lookup_id_random(record_id_random=id_random, table_name='contact')
if values['contact_id_random']:
return redis_lookup_id_random(record_id_random=values['contact_id_random'], table_name='contact')
return None return None
@validator('organization_id', always=True) @validator('organization_id', always=True)
def organization_id_lookup(cls, v, values, **kwargs): def organization_id_lookup(cls, v, values, **kwargs):
log.setLevel(logging.WARNING) if isinstance(v, int) and v > 0: return v
log.debug(locals()) elif id_random := values.get('organization_id_random'):
return redis_lookup_id_random(record_id_random=id_random, table_name='organization')
if values['organization_id_random']:
return redis_lookup_id_random(record_id_random=values['organization_id_random'], table_name='organization')
return None return None
@validator('thumbnail_hosted_file_id', always=True) @validator('thumbnail_hosted_file_id', always=True)

7
app/models/response_models.py
View File

@@ -1,4 +1,3 @@
from __future__ import annotations
import datetime, hashlib, logging, os, pytz, redis, secrets import datetime, hashlib, logging, os, pytz, redis, secrets
from typing import Dict, List, Optional, Set, Union from typing import Dict, List, Optional, Set, Union
@@ -56,6 +55,12 @@ def mk_resp(
elif isinstance(data, list): elif isinstance(data, list):
log.info('Data type is a list') log.info('Data type is a list')
data_out = data data_out = data
elif isinstance(data, int):
log.info('Data type is an int')
data_out = { 'result': data }
elif isinstance(data, str):
log.info('Data type is a str')
data_out = { 'result': data }
else: # Assuming it is still and object. This should be improved. Example model type: "<class 'app.models.account_models.Account_Base'>" else: # Assuming it is still and object. This should be improved. Example model type: "<class 'app.models.account_models.Account_Base'>"
log.info('Data type is other') log.info('Data type is other')
data_out = data.dict(by_alias=by_alias, exclude_unset=exclude_unset) data_out = data.dict(by_alias=by_alias, exclude_unset=exclude_unset)

176
app/routers/e_impexium.py
View File

@@ -12,6 +12,8 @@ from app.routers.api_crud import delete_obj_template, get_obj_template, get_obj_
from app.methods.e_impexium_methods import get_custom_fields, get_events, get_event_registrants from app.methods.e_impexium_methods import get_custom_fields, get_events, get_event_registrants
from app.methods.event_person_methods import create_event_person_obj, create_update_event_person_obj_v4, get_event_person_rec_list, load_event_person_obj, update_event_person_obj, update_event_person_obj_v3
from app.models.event_badge_models import Event_Badge_Base from app.models.event_badge_models import Event_Badge_Base
from app.models.response_models import Resp_Body_Base, mk_resp from app.models.response_models import Resp_Body_Base, mk_resp
@@ -23,94 +25,147 @@ router = APIRouter()
# Updated 2021-10-07 # Updated 2021-10-07
@router.get('/event/{e_impexium_event_id}/import_reg', response_model=Resp_Body_Base) @router.get('/event/{e_impexium_event_id}/import_reg', response_model=Resp_Body_Base)
async def event_import_reg( async def event_import_reg(
e_impexium_event_id: str = Query(..., min_length=11, max_length=22), # For ISHLT: 41V_2 (2021-04) e_impexium_event_id: str = Query(..., min_length=11, max_length=22), # For ISHLT: 42_AM (2022-04); EX22_AM (2022-04); 41V_2 (2021-04)
page: int = 1, # 250 per page from Impexium page: int = 0, # 250 per page from Impexium
event_id: str = Query(..., min_length=11, max_length=22), # For ISHLT: ZDzTBlevhZs (2022-04) event_id: str = Query(..., min_length=11, max_length=22), # For ISHLT: ZDzTBlevhZs (2022-04)
x_account_id: str = Header(...), # For ISHLT: d8TqXqf1EOg # Account ID For ISHLT: d8TqXqf1EOg
response: Response = Response, commons: Common_Route_Params = Depends(common_route_params),
): ):
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL log.setLevel(logging.INFO) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals()) log.debug(locals())
log.info('Starting Impexium event registration import...') log.info('Starting Impexium event registration import...')
if account_id := redis_lookup_id_random(record_id_random=x_account_id, table_name='account'): pass account_id = commons.x_account_id
else: return mk_resp(data=None, status_code=404, status_message=f'The Account ID was not found. Account ID: {account_id}', response=response)
if event_id := redis_lookup_id_random(record_id_random=event_id, table_name='event'): pass 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, status_message=f'The Event ID was not found. Event ID: {event_id}', response=response) else: return mk_resp(data=None, status_code=404, status_message=f'The Event ID was not found. Event ID: {event_id}', response=commons.response)
event_registrant_li = get_event_registrants(event_code=e_impexium_event_id, page=page) event_registrant_li = get_event_registrants(event_code=e_impexium_event_id, page=page)
log.debug(type(event_registrant_li)) log.debug(type(event_registrant_li))
event_badge_li = [] # return mk_resp(data=len(event_registrant_li), status_code=501, response=commons.response) # Not Implemented
loop_count = 0
loop_limit = 2500 # 200
event_person_li = []
for event_registrant in event_registrant_li: for event_registrant in event_registrant_li:
if loop_count > loop_limit: break
log.info(f"Record Num: {event_registrant.get('recordNumber')}, Registration Num: {event_registrant.get('registrationNumber')}, Registrant Type Code: {event_registrant.get('registrantTypeCode')}") log.info(f"Record Num: {event_registrant.get('recordNumber')}, Registration Num: {event_registrant.get('registrationNumber')}, Registrant Type Code: {event_registrant.get('registrantTypeCode')}")
log.debug(event_registrant) log.debug(event_registrant)
event_badge_data = {}
# event_badge_data['account_id'] = account_id
event_badge_data['event_id'] = event_id
# Creating an external ID from the Impexium record number and registration number. Very explicitly converting each number to a string before joining, just in case. # Creating an external ID from the Impexium record number and registration number. Very explicitly converting each number to a string before joining, just in case.
external_id = str(event_registrant.get('recordNumber'))+':'+str(event_registrant.get('registrationNumber')) external_id = str(event_registrant.get('recordNumber'))+':'+str(event_registrant.get('registrationNumber'))
event_person_data = {}
event_person_data['account_id'] = account_id
event_person_data['event_id'] = event_id
event_person_data['external_id'] = external_id
event_person_profile_data = {}
event_person_profile_data['pronouns'] = event_registrant.get('gender')
event_person_profile_data['informal_name'] = event_registrant.get('preferredFirstName')
event_person_profile_data['title_names'] = event_registrant.get('prefix')
event_person_profile_data['given_name'] = event_registrant.get('firstName')
event_person_profile_data['middle_name'] = event_registrant.get('middleName')
event_person_profile_data['family_name'] = event_registrant.get('lastName')
event_person_profile_data['designations'] = event_registrant.get('suffix')
event_person_profile_data['professional_title'] = event_registrant.get('title')
event_person_profile_data['display_name'] = event_registrant.get('badgeName')
event_person_profile_data['affiliations'] = event_registrant.get('badgeOrganization')
event_person_data['event_person_profile'] = {}
event_person_data['event_person_profile'] = event_person_profile_data
event_badge_data = {}
event_badge_data['event_id'] = event_id
event_badge_data['external_id'] = external_id event_badge_data['external_id'] = external_id
sql_select_event_badge = f""" event_badge_data['badge_type_code'] = event_registrant.get('registrantTypeCode') # Using this as the badge_type
SELECT id AS event_badge_id, id_random AS event_badge_id_random, external_id AS event_badge_external_id event_badge_data['badge_type'] = event_registrant.get('registrantTypeName')
FROM `event_badge` AS `event_badge` event_badge_data['member_type_code'] = event_registrant.get('customerType') # Using this as the member_type
WHERE event_badge.event_id = :event_id event_badge_data['registration_type_code'] = event_registrant.get('registrantTypeCode')
AND event_badge.external_id = :external_id event_badge_data['registration_type'] = event_registrant.get('registrantTypeName')
event_badge_data['pronouns'] = event_registrant.get('gender')
event_badge_data['informal_name'] = event_registrant.get('preferredFirstName')
event_badge_data['title_names'] = event_registrant.get('prefix')
event_badge_data['given_name'] = event_registrant.get('firstName')
event_badge_data['middle_name'] = event_registrant.get('middleName')
event_badge_data['family_name'] = event_registrant.get('lastName')
event_badge_data['designations'] = event_registrant.get('suffix')
event_badge_data['professional_title'] = event_registrant.get('title')
event_badge_data['display_name'] = event_registrant.get('badgeName')
event_badge_data['affiliations'] = event_registrant.get('badgeOrganization')
event_badge_data['city'] = event_registrant.get('badgeCity')
event_badge_data['state_province'] = event_registrant.get('badgeState')
event_person_data['event_badge'] = {}
event_person_data['event_badge'] = event_badge_data
sql_select_event_person = f"""
SELECT id AS event_person_id, id_random AS event_person_id_random, external_id AS event_person_external_id, event_badge_id AS event_badge_id, event_person_profile_id AS event_person_profile_id
FROM `event_person` AS `event_person`
WHERE event_person.event_id = :event_id
AND event_person.external_id = :external_id
/*LIMIT 1*/; /*LIMIT 1*/;
""" """
if event_badge_result := sql_select(sql=sql_select_event_badge, data=event_badge_data): if event_person_result := sql_select(sql=sql_select_event_person, data=event_person_data):
log.info('Found Event Badge') if isinstance(event_person_result, list):
if isinstance(event_badge_result, list): log.error(f'Found more than one Event Person with the same External ID. Count: {len(event_person_result)}')
log.error(f'Found more than one Event Badge with the same External ID. Count: {len(event_badge_result)}')
break break
return False # return False
else: else:
log.info('Update record?') event_person_id = event_person_result.get('event_person_id')
event_badge_id = event_person_result.get('event_badge_id')
event_person_profile_id = event_person_result.get('event_person_profile_id')
log.info(f'Found Event Person. Updating existing... Event Person ID: {event_person_id}')
if create_event_person_obj_result := create_update_event_person_obj_v4(
event_person_dict_obj = event_person_data,
event_person_id = event_person_id,
account_id = account_id,
event_id = event_id,
event_badge_id = event_badge_id,
event_person_profile_id = event_person_profile_id,
# create_sub_obj = create_sub_obj,
# fail_any = fail_any,
# return_outline = False,
):
event_person_id = create_event_person_obj_result
log.warning(f'Event Person updated. Event Person ID: {event_person_id}')
else:
log.warning(f'Event Person not updated. Event Person ID: {event_person_id}')
log.debug(event_badge_obj_in_result)
break
# return False
else: else:
log.info('No Event Badge found') log.info('No Event Person found. Creating new...')
log.info('Insert record')
event_badge_data['badge_type_code'] = event_registrant.get('registrantTypeCode') # Using this as the badge_type if create_event_person_obj_result := create_update_event_person_obj_v4(
event_badge_data['badge_type'] = event_registrant.get('registrantTypeName') event_person_dict_obj = event_person_data,
event_badge_data['member_type_code'] = event_registrant.get('customerType') # Using this as the member_type account_id = account_id,
event_badge_data['registration_type_code'] = event_registrant.get('registrantTypeCode') event_id = event_id,
event_badge_data['registration_type'] = event_registrant.get('registrantTypeName') # create_sub_obj = create_sub_obj,
# fail_any = fail_any,
event_badge_data['pronouns'] = event_registrant.get('gender') # return_outline = False,
):
event_badge_data['informal_name'] = event_registrant.get('preferredFirstName') event_person_id = create_event_person_obj_result
event_badge_data['title_names'] = event_registrant.get('prefix') log.warning(f'Event Person created. Event Person ID: {event_person_id}')
event_badge_data['given_name'] = event_registrant.get('firstName')
event_badge_data['middle_name'] = event_registrant.get('middleName')
event_badge_data['family_name'] = event_registrant.get('lastName')
event_badge_data['designations'] = event_registrant.get('suffix')
event_badge_data['professional_title'] = event_registrant.get('title')
event_badge_data['display_name'] = event_registrant.get('badgeName')
event_badge_data['affiliations'] = event_registrant.get('badgeOrganization')
event_badge_data['city'] = event_registrant.get('badgeCity')
event_badge_data['state_province'] = event_registrant.get('badgeState')
if event_badge_obj_in_result := sql_insert(data=event_badge_data, table_name='event_badge', rm_id_random=True, id_random_length=8):
event_badge_id = event_badge_obj_in_result
# event_badge_id = event_badge_obj.get('id') # event_badge_id_random
log.info('Event Badge added. Event Badge ID: {event_badge_id}')
event_badge_data['event_badge_id'] = event_badge_id
else: else:
log.warning(f'Event Badge not created.') log.warning(f'Event Person not created.')
log.debug(event_badge_obj_in_result) log.debug(event_badge_obj_in_result)
break break
return False # return False
# event_badge_data = {} # event_badge_data = {}
# external_id = event_registrant.get('recordNumber')+':'+event_registrant.get('registrationNumber') # external_id = event_registrant.get('recordNumber')+':'+event_registrant.get('registrationNumber')
@@ -144,7 +199,8 @@ async def event_import_reg(
# id, attendedDate, sessions, itemizedCustomFields, guestOfRecordNumber, registrantTypeName, boughtTogetherWith, primaryOrganization, customerType, showInDirectory, addresses, emails, phones # id, attendedDate, sessions, itemizedCustomFields, guestOfRecordNumber, registrantTypeName, boughtTogetherWith, primaryOrganization, customerType, showInDirectory, addresses, emails, phones
event_badge_li.append(event_badge_data) event_person_li.append(event_person_data)
loop_count = loop_count + 1
return mk_resp(data=event_badge_li, response=response) return mk_resp(data=event_person_li, response=commons.response)
# ### END ### API Impexium ### event_import_reg() ### # ### END ### API Impexium ### event_import_reg() ###