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

Work on Impexium API and related.

Browse Source
This commit is contained in:
Scott Idem
2022-02-18 17:35:25 -05:00
parent aa09333383
commit 783ff6e9b3
4 changed files with 130 additions and 127 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
app/methods/e_cvent_methods.py
View File

@@ -1,4 +1,3 @@
from __future__ import annotations
import datetime, json, pprint, pytz, random, requests, secrets, string, time
from requests.auth import HTTPBasicAuth
@@ -61,7 +60,7 @@ def get_access_token():
log.debug(f'Status Code: {resp.status_code}')
log.debug(f'Headers: {resp.headers}')
log.debug(f'Encoding: {resp.encoding}')
# log.debug(f'Encoding: {resp.encoding}')
log.debug(f'JSON: {resp.json()}')
# log.debug('Text:', resp.text)
@@ -171,7 +170,7 @@ def get_contact_custom_field_list(api):
log.debug('Status Code:', resp.status_code)
log.debug('Headers:', resp.headers)
log.debug('Encoding:', resp.encoding)
# log.debug('Encoding:', resp.encoding)
log.debug('JSON:', resp.json())
# log.debug('Text:', resp.text)

238
app/methods/e_impexium_methods.py
View File

@@ -1,11 +1,11 @@
from __future__ import annotations
import datetime, pprint, random, requests
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
from app.lib_general import log, logging, logger_reset
headers = { 'Content-Type': 'application/json;charset=UTF-8' }
@@ -17,60 +17,66 @@ app['name'] = 'IshltOneSkyITLIVE'
app['key'] = '98yp4fa57mJX6nU4'
app['id'] = 'IshltOneSkyITLIVE'
app['password'] = '98yp4fa57mJX6nU4'
print('App data', app)
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' }
# ### BEGIN ### API External Impexium Methods ### get_access_token() ###
# Updated 2022-02-18
@logger_reset
def get_access_token(api, app):
print('***** **** *** ** * ** *** **** *****')
print('*** Get Web API End Point and Access Token ***')
print('***** **** *** ** * ** *** **** *****')
log.setLevel(logging.INFO) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
log.debug(f'API data:\n{api}')
endpoint = '/WebApiUrl'
uri = api['base_url']+endpoint
data = { 'AppName': app['name'], 'AppKey': app['key'] }
r = requests.post(url=uri, json=data, headers=api['headers'])
resp = requests.post(url=uri, json=data, headers=api['headers'])
print('**************************')
print('Status Code:', r.status_code)
print('Headers:', r.headers)
# print('Encoding:', r.encoding)
print('JSON:', r.json())
# print('Text:', r.text)
print('**************************')
log.debug('Status Code:', resp.status_code)
log.debug('Headers:', resp.headers)
# log.debug('Encoding:', resp.encoding)
log.debug('JSON:', resp.json())
# log.debug('Text:', resp.text)
response_data = r.json()
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']
log.debug(api)
return api
# ### END ### API External Impexium Methods ### get_access_token() ###
# ### BEGIN ### API External Impexium Methods ### authenticate() ###
# Updated 2022-02-18
@logger_reset
def authenticate(api, app):
print('***** **** *** ** * ** *** **** *****')
print('*** Authenticate App, and get app token and user token (SSO Token) ***')
print('***** **** *** ** * ** *** **** *****')
log.setLevel(logging.INFO) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
data = { 'AppId': app['id'], 'AppPassword': app['password'], 'AppUserEmail': app['user_email'], 'AppUserPassword': app['user_password'] }
r = requests.post(url=api['auth_uri'], json=data, headers=api['headers'])
resp = requests.post(url=api['auth_uri'], json=data, headers=api['headers'])
print('**************************')
print('Status Code:', r.status_code)
print('Headers:', r.headers)
# print('Encoding:', r.encoding)
print('JSON:', r.json())
# print('Text:', r.text)
print('**************************')
log.debug('Status Code:', resp.status_code)
log.debug('Headers:', resp.headers)
# log.debug('Encoding:', resp.encoding)
log.debug('JSON:', resp.json())
# log.debug('Text:', resp.text)
response_data = r.json()
response_data = resp.json()
api['headers']['AppToken'] = response_data['appToken']
api['headers']['UserToken'] = response_data['userToken']
@@ -80,7 +86,10 @@ def authenticate(api, app):
# app['sso_token'] = response_data['ssoToken']
# app['app_configuration'] = response_data['appConfiguration'] # allowGoogleLogin=False, allowLinkedInLogin=False
log.debug(api)
return api
# ### END ### API External Impexium Methods ### authenticate() ###
api = get_access_token(api=api, app=app)
@@ -90,10 +99,12 @@ api = authenticate(api=api, app=app)
log.debug(api)
# ### BEGIN ### API External Impexium Methods ### get_custom_fields() ###
# Updated 2022-02-18
@logger_reset
def get_custom_fields(api, name=None, page=1):
print('***** **** *** ** * ** *** **** *****')
print('*** Get Custom Fields ***')
print('***** **** *** ** * ** *** **** *****')
log.setLevel(logging.INFO) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
endpoint = f'/Setup/customfields/{page}'
uri = api['base_url']+endpoint
@@ -102,31 +113,28 @@ def get_custom_fields(api, name=None, page=1):
params = {}
r = requests.get(url=uri, params=params, headers=api['headers'])
resp = requests.get(url=uri, params=params, headers=api['headers'])
print('**************************')
print('Status Code:', r.status_code)
# print('Headers:', r.headers)
# print('Encoding:', r.encoding)
print('JSON:', r.json())
# print('Text:', r.text)
print('**************************')
log.debug('Status Code:', resp.status_code)
log.debug('Headers:', resp.headers)
# log.debug('Encoding:', resp.encoding)
log.debug('JSON:', resp.json())
# log.debug('Text:', resp.text)
response_data = r.json()
# pp = pprint.PrettyPrinter(indent=2)
# pp.pprint(response_data)
# print('**************************')
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):
print('***** **** *** ** * ** *** **** *****')
print('*** Get Events ***')
print('***** **** *** ** * ** *** **** *****')
log.setLevel(logging.INFO) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
endpoint = f'/Events/All/{page}'
uri = api['base_url']+endpoint
@@ -135,17 +143,15 @@ def get_events(page=1):
params = {}
r = requests.get(url=uri, params=params, headers=api['headers'])
resp = requests.get(url=uri, params=params, headers=api['headers'])
print('**************************')
print('Status Code:', r.status_code)
# print('Headers:', r.headers)
# print('Encoding:', r.encoding)
print('JSON:', r.json())
# print('Text:', r.text)
print('**************************')
log.debug('Status Code:', resp.status_code)
log.debug('Headers:', resp.headers)
# log.debug('Encoding:', resp.encoding)
log.debug('JSON:', resp.json())
# log.debug('Text:', resp.text)
response_data = r.json()
response_data = resp.json()
# pp = pprint.PrettyPrinter(indent=2)
# pp.pprint(response_data)
@@ -154,35 +160,33 @@ def get_events(page=1):
event_li = response_data
return event_li
# ### END ### API External Impexium Methods ### get_events() ###
# ### BEGIN ### API Impexium Methods ### get_event_registrants() ###
# ### BEGIN ### API External Impexium Methods ### get_event_registrants() ###
# Updated 2021-10-07
def get_event_registrants(
event_code: str,
page: int = 1,
return_all: bool = False
):
print('***** **** *** ** * ** *** **** *****')
print('*** Get Event Registrants ***')
print('***** **** *** ** * ** *** **** *****')
log.setLevel(logging.INFO) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
endpoint = f'/Events/{event_code}/Registrations/{page}'
uri = api['base_url']+endpoint
params = {}
r = requests.get(url=uri, params=params, headers=api['headers'])
resp = requests.get(url=uri, params=params, headers=api['headers'])
print('**************************')
print('Status Code:', r.status_code)
# print('Headers:', r.headers)
# print('Encoding:', r.encoding)
# print('JSON:', r.json())
# print('Text:', r.text)
print('**************************')
log.debug('Status Code:', resp.status_code)
log.debug('Headers:', resp.headers)
# log.debug('Encoding:', resp.encoding)
log.debug('JSON:', resp.json())
# log.debug('Text:', resp.text)
response_data = r.json()
response_data = resp.json()
# pp = pprint.PrettyPrinter(indent=2)
# pp.pprint(response_data)
@@ -197,27 +201,26 @@ def get_event_registrants(
# ### END ### API Impexium Methods ### get_event_registrants() ###
# ### BEGIN ### API External Impexium Methods ### get_individual_registrations() ###
# Updated 2022-02-18
def get_individual_registrations(api, record_number, event_code=None, page=1):
print('***** **** *** ** * ** *** **** *****')
print('*** Get Individual Registrations ***')
print('***** **** *** ** * ** *** **** *****')
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 }
r = requests.get(url=uri, params=params, headers=api['headers'])
resp = requests.get(url=uri, params=params, headers=api['headers'])
print('**************************')
print('Status Code:', r.status_code)
# print('Headers:', r.headers)
# print('Encoding:', r.encoding)
# print('JSON:', r.json())
# print('Text:', r.text)
print('**************************')
log.debug('Status Code:', resp.status_code)
log.debug('Headers:', resp.headers)
# log.debug('Encoding:', resp.encoding)
log.debug('JSON:', resp.json())
# log.debug('Text:', resp.text)
response_data = r.json()
response_data = resp.json()
# pp = pprint.PrettyPrinter(indent=2)
# pp.pprint(response_data)
@@ -226,29 +229,29 @@ def get_individual_registrations(api, record_number, event_code=None, page=1):
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
def get_individual_purchases(api, record_number, product_code=None, purchased_since=None, product_category_code=None, page=1):
print('***** **** *** ** * ** *** **** *****')
print('*** Get Individual Purchases ***')
print('***** **** *** ** * ** *** **** *****')
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 }
r = requests.get(url=uri, params=params, headers=api['headers'])
resp = requests.get(url=uri, params=params, headers=api['headers'])
print('**************************')
print('Status Code:', r.status_code)
# print('Headers:', r.headers)
# print('Encoding:', r.encoding)
print('JSON:', r.json())
# print('Text:', r.text)
print('**************************')
log.debug('Status Code:', resp.status_code)
log.debug('Headers:', resp.headers)
# log.debug('Encoding:', resp.encoding)
log.debug('JSON:', resp.json())
# log.debug('Text:', resp.text)
response_data = r.json()
response_data = resp.json()
# pp = pprint.PrettyPrinter(indent=2)
# pp.pprint(response_data)
@@ -257,29 +260,29 @@ def get_individual_purchases(api, record_number, product_code=None, purchased_si
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-02-18
def get_individual_custom_fields(api, individual_id):
print('***** **** *** ** * ** *** **** *****')
print('*** Get Individual Custom Fields ***')
print('***** **** *** ** * ** *** **** *****')
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 = { }
r = requests.get(url=uri, params=params, headers=api['headers'])
resp = requests.get(url=uri, params=params, headers=api['headers'])
print('**************************')
print('Status Code:', r.status_code)
# print('Headers:', r.headers)
# print('Encoding:', r.encoding)
# print('JSON:', r.json())
# print('Text:', r.text)
print('**************************')
log.debug('Status Code:', resp.status_code)
log.debug('Headers:', resp.headers)
# log.debug('Encoding:', resp.encoding)
log.debug('JSON:', resp.json())
# log.debug('Text:', resp.text)
response_data = r.json()
response_data = resp.json()
# pp = pprint.PrettyPrinter(indent=2)
# pp.pprint(response_data)
@@ -288,12 +291,14 @@ def get_individual_custom_fields(api, individual_id):
individual_custom_field_li = response_data
return individual_custom_field_li
# ### END ### API External Impexium Methods ### get_individual_custom_fields() ###
# ### BEGIN ### API External Impexium Methods ### get_individual_orders_open() ###
# Updated 2022-02-18
def get_individual_orders_open(api, record_number, include_line_items=False, from_date=None, to_date=None, page=1):
print('***** **** *** ** * ** *** **** *****')
print('*** Get Individual Orders Open ***')
print('***** **** *** ** * ** *** **** *****')
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
@@ -301,17 +306,15 @@ def get_individual_orders_open(api, record_number, include_line_items=False, fro
params = { 'includeLineItems': include_line_items, 'fromDate': from_date, 'toDate': to_date }
r = requests.get(url=uri, params=params, headers=api['headers'])
resp = requests.get(url=uri, params=params, headers=api['headers'])
print('**************************')
print('Status Code:', r.status_code)
print('Headers:', r.headers)
# print('Encoding:', r.encoding)
print('JSON:', r.json())
print('Text:', r.text)
print('**************************')
log.debug('Status Code:', resp.status_code)
log.debug('Headers:', resp.headers)
# log.debug('Encoding:', resp.encoding)
log.debug('JSON:', resp.json())
# log.debug('Text:', resp.text)
response_data = r.json()
response_data = resp.json()
# pp = pprint.PrettyPrinter(indent=2)
# pp.pprint(response_data)
@@ -320,3 +323,4 @@ def get_individual_orders_open(api, record_number, include_line_items=False, fro
individual_orders_open_li = response_data
return individual_orders_open_li
# ### END ### API External Impexium Methods ### get_individual_orders_open() ###