513 lines
18 KiB
Python
513 lines
18 KiB
Python
import datetime, json, pprint, pytz, random, requests, string, time
|
|
|
|
from typing import Dict, List, Optional, Set, Union
|
|
from pydantic import BaseModel, EmailStr, Field, PrivateAttr, ValidationError, validator
|
|
|
|
from app.db_sql import redis_lookup_id_random, sql_insert, sql_select, sql_update
|
|
from app.lib_general import log, logging, logger_reset
|
|
|
|
|
|
headers = { 'Content-Type': 'application/json;charset=UTF-8' }
|
|
|
|
app = {}
|
|
app['user_email'] = 'oneskyit_integration@oneskyit.com'
|
|
app['user_password'] = 'dAPpHDE6d5vLHjsk'
|
|
|
|
app['name'] = 'IshltOneSkyITLIVE'
|
|
app['key'] = '98yp4fa57mJX6nU4'
|
|
app['id'] = 'IshltOneSkyITLIVE'
|
|
app['password'] = '98yp4fa57mJX6nU4'
|
|
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
|
|
api['access_token_datetime'] = None # This is NOT generated by Impexium
|
|
api['app_user_token_datetime'] = None # This is NOT generated by Impexium
|
|
|
|
|
|
# ### BEGIN ### API External Impexium Methods ### get_access_token() ###
|
|
# Updated 2022-03-23
|
|
@logger_reset
|
|
def get_access_token():
|
|
log.setLevel(logging.INFO) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
|
|
log.debug(locals())
|
|
|
|
log.debug(f'App data:\n{app}')
|
|
log.debug(f'API data:\n{api}')
|
|
|
|
endpoint = '/WebApiUrl'
|
|
uri = api['base_url']+endpoint
|
|
|
|
data = { 'AppName': app['name'], 'AppKey': app['key'] }
|
|
|
|
resp = requests.post(url=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.status_code == 200 and resp.json():
|
|
log.info('Got access token. Setting API header and related...')
|
|
log.debug(f'JSON: {resp.json()}')
|
|
else:
|
|
log.warning('No JSON data found')
|
|
return False
|
|
# log.debug('Text:', resp.text)
|
|
|
|
response_data = resp.json()
|
|
log.debug(json.dumps(response_data, indent=2, default=str))
|
|
|
|
api['access_token'] = response_data['accessToken']
|
|
api['headers']['AccessToken'] = response_data['accessToken']
|
|
api['auth_uri'] = response_data['uri']
|
|
|
|
api['access_token_datetime'] = datetime.datetime.utcnow() # This is NOT generated by Impexium
|
|
|
|
log.info('Finished setting API access token information.')
|
|
log.debug(api)
|
|
|
|
return True
|
|
# ### END ### API External Impexium Methods ### get_access_token() ###
|
|
|
|
|
|
# ### BEGIN ### API External Impexium Methods ### authenticate() ###
|
|
# Updated 2022-03-23
|
|
@logger_reset
|
|
def authenticate():
|
|
log.setLevel(logging.INFO) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
|
|
log.debug(locals())
|
|
|
|
if api.get('access_token') and api.get('app_user_token_datetime'):
|
|
log.info(f'Found API Access Token and timestamps: Access={api.get("access_token_datetime")}; App/User={api.get("app_user_token_datetime")}')
|
|
log.debug(api.get('access_token'))
|
|
# log.debug(api.get('access_token_datetime'))
|
|
# log.debug(api.get('app_user_token_datetime'))
|
|
return True
|
|
elif api.get('access_token'): # pass
|
|
log.info('Found API Access Token but no timestamp. Need to authenticate.')
|
|
log.debug(api.get('access_token'))
|
|
else:
|
|
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}')
|
|
|
|
data = { 'AppId': app['id'], 'AppPassword': app['password'], 'AppUserEmail': app['user_email'], 'AppUserPassword': app['user_password'] }
|
|
|
|
try_request = True
|
|
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}')
|
|
# log.debug('Text:', resp.text)
|
|
|
|
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()
|
|
log.debug(json.dumps(response_data, indent=2, default=str))
|
|
else:
|
|
log.warning('No JSON data found')
|
|
response_data = None
|
|
try_request = False
|
|
elif resp.status_code == 429:
|
|
log.warning('Hit rate limit. Sleeping for .1 seconds...')
|
|
time.sleep(.1)
|
|
try_request = True
|
|
else:
|
|
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']
|
|
|
|
api['app_user_token_datetime'] = datetime.datetime.utcnow() # This is NOT generated by Impexium
|
|
|
|
log.debug(api)
|
|
|
|
return api
|
|
# ### END ### API External Impexium Methods ### authenticate() ###
|
|
|
|
|
|
# ### BEGIN ### API External Impexium Methods ### get_custom_fields() ###
|
|
# Updated 2022-02-18
|
|
@logger_reset
|
|
def get_custom_fields(api, name=None, page=1):
|
|
log.setLevel(logging.INFO) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
|
|
log.debug(locals())
|
|
|
|
endpoint = f'/Setup/customfields/{page}'
|
|
uri = api['base_url']+endpoint
|
|
|
|
print('Endpoint URI', uri)
|
|
|
|
params = {}
|
|
|
|
resp = requests.get(url=uri, params=params, headers=api['headers'])
|
|
|
|
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()
|
|
|
|
custom_field_li = response_data
|
|
|
|
return custom_field_li
|
|
# ### END ### API External Impexium Methods ### get_custom_fields() ###
|
|
|
|
|
|
# ### BEGIN ### API External Impexium Methods ### get_events() ###
|
|
# Updated 2022-02-18
|
|
@logger_reset
|
|
def get_events(page=1):
|
|
log.setLevel(logging.INFO) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
|
|
log.debug(locals())
|
|
|
|
endpoint = f'/Events/All/{page}'
|
|
uri = api['base_url']+endpoint
|
|
|
|
print('Endpoint URI', uri)
|
|
|
|
params = {}
|
|
|
|
resp = requests.get(url=uri, params=params, headers=api['headers'])
|
|
|
|
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()
|
|
|
|
# pp = pprint.PrettyPrinter(indent=2)
|
|
# pp.pprint(response_data)
|
|
# print('**************************')
|
|
|
|
event_li = response_data
|
|
|
|
return event_li
|
|
# ### END ### API External Impexium Methods ### get_events() ###
|
|
|
|
|
|
# ### 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,
|
|
details: bool = False,
|
|
page: int = 0,
|
|
# return_all: bool = False
|
|
):
|
|
log.setLevel(logging.INFO) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
|
|
log.debug(locals())
|
|
|
|
if result := authenticate():
|
|
log.debug(result)
|
|
else:
|
|
return False
|
|
|
|
try_page = True
|
|
|
|
if page > 0: # Only get the specific page
|
|
page_num = page
|
|
max_page = page
|
|
else: # Get all of the pages
|
|
page = 1
|
|
page_num = 0 # Will actually be the first page
|
|
max_page = 15
|
|
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} 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
|
|
params = { 'includeDetails': details }
|
|
if registered_since:
|
|
params['registeredSince'] = registered_since.isoformat()
|
|
log.debug(params)
|
|
|
|
try_request = True
|
|
max_tries = 5
|
|
try_count = 0
|
|
while try_request and try_count <= max_tries:
|
|
try_count = try_count + 1
|
|
|
|
resp = requests.get(url=uri, params=params, headers=api['headers'])
|
|
|
|
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 resp.status_code == 200:
|
|
response_data = resp.json()
|
|
|
|
current_page = response_data.get('pageNumber')
|
|
log.info(f'Impexium response Current Page: {current_page}')
|
|
|
|
impexium_event_registration_list = impexium_event_registration_list + response_data.get('dataList')
|
|
|
|
try_request = False
|
|
elif resp.status_code == 404:
|
|
log.info('No results returned.')
|
|
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:
|
|
log.info('Not trying again')
|
|
try_request = False
|
|
try_page = False
|
|
impexium_event_registration_list = False
|
|
|
|
log.warning('Something may have gone wrong. Setting the API app_user_token_datetime value to None to re-authenticate with Impexium on the next request.')
|
|
api['app_user_token_datetime'] = None # Resetting this just in case the App and or User token expired.
|
|
|
|
return impexium_event_registration_list
|
|
# ### END ### API Impexium Methods ### get_event_registrants() ###
|
|
|
|
|
|
|
|
# ### BEGIN ### API External Impexium Methods ### get_individual_profile() ###
|
|
# Updated 2022-04-22
|
|
# NOTE: Without details the results are very basic. Pretty much just their name.
|
|
# Including details adds the addresses, customFields, emails, memberships, phones, etc
|
|
@logger_reset
|
|
def get_individual_profile(
|
|
individual_id: str,
|
|
details: bool = False,
|
|
page = 1 # page: int = 0,
|
|
):
|
|
log.setLevel(logging.INFO) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
|
|
log.debug(locals())
|
|
|
|
if result := authenticate():
|
|
log.debug(result)
|
|
else:
|
|
return False
|
|
|
|
endpoint = f'/Individuals/Profile/{individual_id}/{page}'
|
|
uri = api['base_url']+endpoint
|
|
params = { 'IncludeDetails': details }
|
|
|
|
impexium_individual_profile = None
|
|
|
|
try_request = True
|
|
max_tries = 5
|
|
try_count = 0
|
|
while try_request and try_count <= max_tries:
|
|
try_count = try_count + 1
|
|
|
|
resp = requests.get(url=uri, params=params, headers=api['headers'])
|
|
|
|
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)
|
|
# log.debug(resp.json())
|
|
|
|
if resp.status_code == 200:
|
|
log.info('Status 200')
|
|
log.debug(resp.json())
|
|
|
|
impexium_individual_profile_raw = resp.json() # .get('data').get('dataList')[0]
|
|
# log.debug(impexium_individual_profile_raw)
|
|
|
|
impexium_individual_profile = impexium_individual_profile_raw.get('dataList')[0]
|
|
log.debug(impexium_individual_profile)
|
|
|
|
try_request = False
|
|
elif resp.status_code == 404:
|
|
log.info('No results returned (status 404)')
|
|
try_request = False
|
|
impexium_individual_profile = None
|
|
elif resp.status_code == 429:
|
|
log.warning('Hit rate limit. Sleeping for .1 seconds...')
|
|
time.sleep(.1)
|
|
try_request = True
|
|
impexium_individual_profile = False
|
|
else:
|
|
log.info('Not trying again')
|
|
try_request = False
|
|
impexium_event_registration_list = False
|
|
|
|
log.warning('Something may have gone wrong. Setting the API app_user_token_datetime value to None to re-authenticate with Impexium on the next request.')
|
|
api['app_user_token_datetime'] = None # Resetting this just in case the App and or User token expired.
|
|
|
|
return impexium_individual_profile
|
|
# ### END ### API External Impexium Methods ### get_individual_profile() ###
|
|
|
|
|
|
|
|
|
|
# ### 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())
|
|
|
|
endpoint = f'/Individuals/{record_number}/Registrations/{page}'
|
|
uri = api['base_url']+endpoint
|
|
|
|
params = { 'eventCode': event_code }
|
|
|
|
resp = requests.get(url=uri, params=params, headers=api['headers'])
|
|
|
|
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()
|
|
|
|
# pp = pprint.PrettyPrinter(indent=2)
|
|
# pp.pprint(response_data)
|
|
# print('**************************')
|
|
|
|
individual_registraion_li = response_data
|
|
|
|
return individual_registraion_li
|
|
# ### END ### API External Impexium Methods ### get_individual_registrations() ###
|
|
|
|
|
|
# ### 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())
|
|
|
|
endpoint = f'/Individuals/{record_number}/Purchases/{page}'
|
|
uri = api['base_url']+endpoint
|
|
|
|
params = { 'productCode': product_code, 'purchasedSince': purchased_since, 'productCategoryCode': product_category_code }
|
|
|
|
resp = requests.get(url=uri, params=params, headers=api['headers'])
|
|
|
|
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()
|
|
|
|
# pp = pprint.PrettyPrinter(indent=2)
|
|
# pp.pprint(response_data)
|
|
# print('**************************')
|
|
|
|
individual_purchase_li = response_data
|
|
|
|
return individual_purchase_li
|
|
# ### END ### API External Impexium Methods ### get_individual_purchases() ###
|
|
|
|
|
|
# ### BEGIN ### API External Impexium Methods ### get_individual_custom_fields() ###
|
|
# 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 = { }
|
|
|
|
impexium_individual_custom_field_list = None
|
|
|
|
try_request = True
|
|
max_tries = 5
|
|
try_count = 0
|
|
while try_request and try_count <= max_tries:
|
|
try_count = try_count + 1
|
|
|
|
resp = requests.get(url=uri, params=params, headers=api['headers'])
|
|
|
|
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 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 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())
|
|
|
|
endpoint = f'/Individuals/{record_number}/Orders/Open/{page}'
|
|
uri = api['base_url']+endpoint
|
|
print(uri)
|
|
|
|
params = { 'includeLineItems': include_line_items, 'fromDate': from_date, 'toDate': to_date }
|
|
|
|
resp = requests.get(url=uri, params=params, headers=api['headers'])
|
|
|
|
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()
|
|
|
|
# pp = pprint.PrettyPrinter(indent=2)
|
|
# pp.pprint(response_data)
|
|
# print('**************************')
|
|
|
|
individual_orders_open_li = response_data
|
|
|
|
return individual_orders_open_li
|
|
# ### END ### API External Impexium Methods ### get_individual_orders_open() ###
|