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

More improvement to Impexium API calls. Should be faster and more robust now.

Browse Source
This commit is contained in:
Scott Idem
2022-03-23 15:38:53 -04:00
parent 7c7bd3c530
commit c625cda845
5 changed files with 120 additions and 55 deletions
Download Patch File Download Diff File Expand all files Collapse all files

101
app/methods/e_impexium_methods.py
View File

@@ -26,10 +26,10 @@ api['access_token'] = None
# ### BEGIN ### API External Impexium Methods ### get_access_token() ###
# Updated 2022-02-18
# Updated 2022-03-23
@logger_reset
def get_access_token():
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(f'App data:\n{app}')
@@ -49,6 +49,7 @@ def get_access_token():
log.debug(f'JSON: {resp.json()}')
else:
log.warning('No JSON data found')
return False
# log.debug('Text:', resp.text)
response_data = resp.json()
@@ -60,21 +61,24 @@ def get_access_token():
log.debug(api)
return api
return True
# ### END ### API External Impexium Methods ### get_access_token() ###
# ### BEGIN ### API External Impexium Methods ### authenticate() ###
# Updated 2022-03-22
# Updated 2022-03-23
@logger_reset
def authenticate():
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.setLevel(logging.INFO) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
if api.get('access_token'): pass
else:
result = get_access_token()
log.debug(result)
log.warning('Access token not found. Calling get_access_token()...')
if result := get_access_token():
log.debug(result)
else:
return False
log.debug(f'App data:\n{app}')
log.debug(f'API data:\n{api}')
@@ -122,7 +126,6 @@ def authenticate():
# ### END ### API External Impexium Methods ### authenticate() ###
# ### BEGIN ### API External Impexium Methods ### get_custom_fields() ###
# Updated 2022-02-18
@logger_reset
@@ -189,6 +192,7 @@ def get_events(page=1):
# ### BEGIN ### API External Impexium Methods ### get_event_registrants() ###
# Updated 2021-10-07
@logger_reset
def get_event_registrants(
event_code: str,
registered_since: datetime.datetime = None,
@@ -196,11 +200,13 @@ def get_event_registrants(
page: int = 0,
# return_all: bool = False
):
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.setLevel(logging.INFO) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
result = authenticate()
log.debug(result)
if result := authenticate():
log.debug(result)
else:
return False
try_page = True
@@ -214,7 +220,7 @@ def get_event_registrants(
impexium_event_registration_list = []
while try_page and page_num <= max_page:
page_num = page_num + 1
log.info(f'Getting page number {page_num}')
log.info(f'Getting page number {page_num} from Impexium... Event Code: {event_code}; Details: {details}; Registered Since: {registered_since}')
endpoint = f'/Events/{event_code}/Registrations/{page_num}'
uri = api['base_url']+endpoint
@@ -241,16 +247,13 @@ def get_event_registrants(
response_data = resp.json()
current_page = response_data.get('pageNumber')
log.info(f'Current Page: {current_page}')
log.info(f'Impexium response Current Page: {current_page}')
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)
try_request = False
elif resp.status_code == 404:
log.info('No results returned.')
try_request = False
try_page = False
elif resp.status_code == 429:
@@ -263,24 +266,13 @@ def get_event_registrants(
try_page = False
impexium_event_registration_list = False
# 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() ###
# ### BEGIN ### API External Impexium Methods ### get_individual_registrations() ###
# Updated 2022-02-18
@logger_reset
def get_individual_registrations(api, record_number, event_code=None, page=1):
log.setLevel(logging.INFO) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
@@ -312,6 +304,7 @@ def get_individual_registrations(api, record_number, event_code=None, page=1):
# ### BEGIN ### API External Impexium Methods ### get_individual_purchases() ###
# Updated 2022-02-18
@logger_reset
def get_individual_purchases(api, record_number, product_code=None, purchased_since=None, product_category_code=None, page=1):
log.setLevel(logging.INFO) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
@@ -342,38 +335,58 @@ def get_individual_purchases(api, record_number, product_code=None, purchased_si
# ### BEGIN ### API External Impexium Methods ### get_individual_custom_fields() ###
# Updated 2022-02-18
def get_individual_custom_fields(api, individual_id):
# Updated 2022-03-23
@logger_reset
def get_individual_custom_fields(
individual_id
):
log.setLevel(logging.INFO) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
endpoint = f'/Individuals/{individual_id}/CustomFields'
uri = api['base_url']+endpoint
params = { }
resp = requests.get(url=uri, params=params, headers=api['headers'])
impexium_individual_custom_field_list = None
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)
try_request = True
max_tries = 5
try_count = 0
while try_request and try_count <= max_tries:
try_count = try_count + 1
response_data = resp.json()
resp = requests.get(url=uri, params=params, headers=api['headers'])
# pp = pprint.PrettyPrinter(indent=2)
# pp.pprint(response_data)
# print('**************************')
log.debug(f'Status Code: {resp.status_code}')
log.debug(f'Headers: {resp.headers}')
# log.debug(f'Encoding: {resp.encoding}')
# log.debug('Text:')
# log.debug(resp.text)
individual_custom_field_li = response_data
if resp.status_code == 200:
log.debug(resp.json)
impexium_individual_custom_field_list = resp.json()
try_request = False
elif resp.status_code == 404:
log.info('No results returned.')
try_request = False
impexium_individual_custom_field_list = []
elif resp.status_code == 429:
log.warning('Hit rate limit. Sleeping for .1 seconds...')
time.sleep(.1)
try_request = True
impexium_individual_custom_field_list = False
else:
try_request = False
impexium_event_registration_list = False
return individual_custom_field_li
return impexium_individual_custom_field_list
# ### END ### API External Impexium Methods ### get_individual_custom_fields() ###
# ### BEGIN ### API External Impexium Methods ### get_individual_orders_open() ###
# Updated 2022-02-18
@logger_reset
def get_individual_orders_open(api, record_number, include_line_items=False, from_date=None, to_date=None, page=1):
log.setLevel(logging.INFO) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())

3
app/methods/event_badge_methods.py
View File

@@ -98,6 +98,7 @@ def get_event_badge_template_id_w_event_id(
# ### BEGIN ### API Event Badge Methods ### create_update_event_badge_obj_v4() ###
# Updated 2022-02-23
@logger_reset
def create_update_event_badge_obj_v4(
event_badge_dict_obj: Event_Badge_Base|dict,
event_badge_id: int|str = None,
@@ -108,7 +109,7 @@ def create_update_event_badge_obj_v4(
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.setLevel(logging.INFO) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
# ### SECTION ### Secondary data validation