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

Improved the main Impexium API calls. Should be faster and more robust now.

Browse Source
This commit is contained in:
Scott Idem
2022-03-22 16:59:06 -04:00
parent e1eeb8990c
commit 7c7bd3c530
2 changed files with 132 additions and 59 deletions
Download Patch File Download Diff File Expand all files Collapse all files

150
app/methods/e_impexium_methods.py
View File

@@ -22,6 +22,7 @@ log.debug('App data', app)
api = {}
api['base_url'] = 'https://public.impexium.com/Api/v1' # or https://ishlt.mpxapi.com:443/api/v1 ??
api['headers'] = { 'Content-Type': 'application/json;charset=UTF-8' }
api['access_token'] = None
# ### BEGIN ### API External Impexium Methods ### get_access_token() ###
@@ -64,14 +65,16 @@ def get_access_token():
# ### BEGIN ### API External Impexium Methods ### authenticate() ###
# Updated 2022-02-18
# Updated 2022-03-22
@logger_reset
def authenticate():
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
result = get_access_token()
log.debug(result)
if api.get('access_token'): pass
else:
result = get_access_token()
log.debug(result)
log.debug(f'App data:\n{app}')
log.debug(f'API data:\n{api}')
@@ -79,30 +82,39 @@ def authenticate():
data = { 'AppId': app['id'], 'AppPassword': app['password'], 'AppUserEmail': app['user_email'], 'AppUserPassword': app['user_password'] }
try_request = True
limit = 0
while try_request and limit < 3:
limit = limit + 1
max_tries = 5
try_count = 0
while try_request and try_count < max_tries:
try_count = try_count + 1
resp = requests.post(url=api['auth_uri'], json=data, headers=api['headers'])
log.debug(f'Status Code: {resp.status_code}')
log.debug(f'Headers: {resp.headers}')
# log.debug(f'Encoding: {resp.encoding}')
if resp.json():
log.debug(f'JSON: {resp.json()}')
else:
log.warning('No JSON data found')
# log.debug('Text:', resp.text)
if resp:
if resp.status_code == 200:
log.info('Got a result from request')
if resp.json():
log.debug(f'JSON: {resp.json()}')
response_data = resp.json()
else:
log.warning('No JSON data found')
response_data = None
try_request = False
response_data = resp.json()
api['headers']['AppToken'] = response_data['appToken']
api['headers']['UserToken'] = response_data['userToken']
api['base_url'] = response_data['uri']
elif resp.status_code == 429:
log.warning('Hit rate limit. Sleeping for .1 seconds...')
time.sleep(.1)
try_request = True
else:
log.warning('Hit rate limit? Sleeping for .5 seconds...')
time.sleep(.5)
log.error('Unexpected result from request')
try_request = False
response_data = False
api['headers']['AppToken'] = response_data['appToken']
api['headers']['UserToken'] = response_data['userToken']
api['base_url'] = response_data['uri']
log.debug(api)
@@ -110,12 +122,6 @@ def authenticate():
# ### END ### API External Impexium Methods ### authenticate() ###
# api = get_access_token(api=api, app=app)
# log.debug(api)
# api = authenticate(api=api, app=app)
# log.debug(api)
# ### BEGIN ### API External Impexium Methods ### get_custom_fields() ###
# Updated 2022-02-18
@@ -185,63 +191,89 @@ def get_events(page=1):
# Updated 2021-10-07
def get_event_registrants(
event_code: str,
registered_since: datetime.datetime = None,
details: bool = False,
page: int = 0,
# 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())
result = authenticate()
log.debug(result)
try_request = True
try_page = True
if page > 0:
loop_count = page
loop_limit = page + 1
else:
if page > 0: # Only get the specific page
page_num = page
max_page = page
else: # Get all of the pages
page = 1
loop_count = 0
loop_limit = 15
# current_page = 0
page_num = 0 # Will actually be the first page
max_page = 15
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}'
while try_page and page_num <= max_page:
page_num = page_num + 1
log.info(f'Getting page number {page_num}')
endpoint = f'/Events/{event_code}/Registrations/{page_num}'
uri = api['base_url']+endpoint
params = { 'includeDetails': details }
if registered_since:
params['registeredSince'] = registered_since.isoformat()
log.debug(params)
if resp := requests.get(url=uri, params=params, headers=api['headers']): pass
else: break
try_request = True
max_tries = 5
try_count = 0
while try_request and try_count <= max_tries:
try_count = try_count + 1
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)
resp = requests.get(url=uri, params=params, headers=api['headers'])
response_data = resp.json()
current_page = response_data.get('pageNumber')
log.info(f'Current Page: {current_page}')
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)
if current_page >= 1:
impexium_event_registration_list = impexium_event_registration_list + response_data.get('dataList')
if resp.status_code == 200:
response_data = resp.json()
page = current_page + 1
current_page = response_data.get('pageNumber')
log.info(f'Current Page: {current_page}')
log.warning('Avoiding rate limit. Sleeping for .1 seconds...')
time.sleep(.1)
else:
try_request = False
impexium_event_registration_list = impexium_event_registration_list + response_data.get('dataList')
loop_count = loop_count + 1
# page = current_page + 1
# log.warning('Avoiding rate limit. Sleeping for .1 seconds...')
# time.sleep(.1)
try_request = False
elif resp.status_code == 404:
try_request = False
try_page = False
elif resp.status_code == 429:
log.warning('Hit rate limit. Sleeping for .1 seconds...')
time.sleep(.1)
try_request = True
response_data = False
else:
try_request = False
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() ###

41
app/routers/e_impexium.py
View File

@@ -26,6 +26,7 @@ router = APIRouter()
@router.get('/event/{e_impexium_event_id}/import_reg', response_model=Resp_Body_Base)
async def event_import_reg(
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)
registered_since: datetime.datetime = None, # datetime.datetime.now() + datetime.timedelta(seconds=120)
details: bool = True,
page: int = 0, # 250 per page from Impexium
event_id: str = Query(..., min_length=11, max_length=22), # For ISHLT: ZDzTBlevhZs (2022-04)
@@ -335,3 +336,43 @@ async def event_import_reg(
return mk_resp(data=event_person_li, response=commons.response)
# ### END ### API Impexium ### event_import_reg() ###
# ### BEGIN ### API Impexium ### testing() ###
# Updated 2022-03-22
@router.get('/event/{e_impexium_event_id}/testing', response_model=Resp_Body_Base)
async def testing(
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)
registered_since: datetime.datetime = None, # datetime.datetime.now() + datetime.timedelta(seconds=120)
details: bool = True,
page: int = 0, # 250 per page from Impexium
event_id: str = Query(..., min_length=11, max_length=22), # For ISHLT: ZDzTBlevhZs (2022-04)
# Account ID For ISHLT: d8TqXqf1EOg
commons: Common_Route_Params = Depends(common_route_params),
):
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
log.info('Starting Impexium event registration import...')
account_id = commons.x_account_id
event_id = 'ZDzTBlevhZs' # ISHLT 2022-04 Boston
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=commons.response)
event_registrant_li = get_event_registrants(
event_code = e_impexium_event_id,
registered_since = registered_since,
details = details,
page = page
)
log.debug(type(event_registrant_li))
if event_registrant_li:
return mk_resp(data=event_registrant_li, status_code=200, response=commons.response)
# return mk_resp(data=len(event_registrant_li), status_code=200, response=commons.response)
else:
return mk_resp(data=event_registrant_li, status_code=400, response=commons.response, status_message='Possible bad request???') # Bad Request
# ### END ### API Impexium ### testing() ###