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

Working on Cvent API related

Browse Source
This commit is contained in:
Scott Idem
2022-01-31 19:37:02 -05:00
parent 1d39a01ded
commit 779bbb2f82
9 changed files with 404 additions and 111 deletions
Download Patch File Download Diff File Expand all files Collapse all files

201
app/routers/e_cvent.py
View File

@@ -3,15 +3,18 @@ from fastapi import APIRouter, Body, Depends, Header, HTTPException, Query, Resp
from pydantic import BaseModel, EmailStr, Field
from typing import Dict, List, Optional, Set, Union
from app.lib_general import log, logging, secure_hash_string
from app.lib_general import log, logging, secure_hash_string, verify_secure_hash_string, common_route_params, Common_Route_Params
from app.config import settings
from app.db_sql import sql_insert, sql_update, sql_insert_or_update, sql_select, sql_delete, redis_lookup_id_random
from app.routers.api_crud import delete_obj_template, get_obj_template, get_obj_li_template, patch_obj_template, post_obj_template
from app.methods.e_cvent_methods import get_customer_list, get_customer, get_custom_fields
from app.methods.e_cvent_methods import get_access_token, get_contact_custom_field_list, get_contact_list, get_contact_id
from app.methods.person_methods import get_person_rec_list, get_person_rec_w_external_id, load_person_obj, update_person_kiss
from app.methods.membership_person_methods import create_membership_person_obj, update_membership_person_obj
from app.models.response_models import Resp_Body_Base, mk_resp
from app.models.person_models import Person_Base
router = APIRouter()
@@ -21,19 +24,38 @@ router = APIRouter()
# Updated 2022-01-28
@router.get('/person/{e_person_id}', response_model=Resp_Body_Base)
async def get_person(
e_person_id: str = Query(..., min_length=36, max_length=36), # UUID v4
e_person_id: str = Query(..., min_length=36, max_length=36), # UUID v4; actually the Cvent Contact UUID for a person
commons: Common_Route_Params = Depends(common_route_params),
):
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
if cvent_customer_obj_result := get_customer(customer_id=e_person_id):
customer_obj = cvent_customer_obj_result
# current_datetime = datetime.datetime.now()
# log.debug(type(current_datetime))
# log.debug(current_datetime)
# expires_in = 3600
# log.debug(type(expires_in))
# log.debug(expires_in)
# future_datetime = current_datetime + datetime.timedelta(seconds=3600)
# log.debug(type(future_datetime))
# log.debug(future_datetime)
# return mk_resp(data=False, status_code=501, response=commons.response, status_message='Testing...')
if cvent_contact_obj_result := get_contact_id(contact_id=e_person_id):
cvent_contact_obj = cvent_contact_obj_result
else:
log.info(f'Cvent contact not found with ID: {e_person_id}')
return mk_resp(data=None, status_code=404, response=commons.response) # Not Found
account_id = commons.x_account_id
# return mk_resp(data=False, status_code=501, response=commons.response, status_message='Testing...')
# Important contact fields:
# id (UUID v4)
# type { id, name } for contact type (membership type)
@@ -56,50 +78,57 @@ async def get_person(
# homeAddress { address1, address2, address3, city, country, countryCode, postalCode, region, regionCode }
person_data = {}
if custom_field_list := customer_obj.get('customFields'): # List
if custom_field_list := cvent_contact_obj.get('customFields'): # List
for custom_field in custom_field_list:
# Get the External ID created by OSIT
if custom_field.get('id') == '609ab766-7d79-4a9d-a72c-f126412659ee':
person_data['external_id'] = custom_field.get('value')
person_data['external_id'] = custom_field.get('value')[0]
# person_data['pronouns'] = ???
person_data['informal_name'] = customer_obj.get('nickname')
person_data['title_names'] = customer_obj.get('prefix')
person_data['given_name'] = customer_obj.get('firstName')
person_data['middle_name'] = customer_obj.get('middleName')
person_data['family_name'] = customer_obj.get('lastName')
person_data['designations'] = customer_obj.get('designation')
person_data['professional_title'] = customer_obj.get('title')
person_data['informal_name'] = cvent_contact_obj.get('nickname')
person_data['title_names'] = cvent_contact_obj.get('prefix')
person_data['given_name'] = cvent_contact_obj.get('firstName')
person_data['middle_name'] = cvent_contact_obj.get('middleName')
person_data['family_name'] = cvent_contact_obj.get('lastName')
person_data['designations'] = cvent_contact_obj.get('designation')
person_data['professional_title'] = cvent_contact_obj.get('title')
person_data['affiliations'] = customer_obj.get('company')
person_data['affiliations'] = cvent_contact_obj.get('company')
log.debug(person_data)
contact_data = {}
contact_data['email'] = customer_obj.get('email')
contact_data['cc_email'] = customer_obj.get('ccEmail')
contact_data['phone_mobile'] = customer_obj.get('mobilePhone')
contact_data['phone_home'] = customer_obj.get('homePhone')
contact_data['phone_office'] = customer_obj.get('workPhone')
contact_data['email'] = cvent_contact_obj.get('email')
contact_data['cc_email'] = cvent_contact_obj.get('ccEmail')
contact_data['phone_mobile'] = cvent_contact_obj.get('mobilePhone')
contact_data['phone_home'] = cvent_contact_obj.get('homePhone')
contact_data['phone_office'] = cvent_contact_obj.get('workPhone')
log.debug(contact_data)
address_data = {}
address_data['line_1'] = customer_obj.get('homeAddress').get('address1')
address_data['line_2'] = customer_obj.get('homeAddress').get('address2')
address_data['line_3'] = customer_obj.get('homeAddress').get('address3')
address_data['city'] = customer_obj.get('homeAddress').get('city')
address_data['state_province'] = customer_obj.get('homeAddress').get('region')
country_subdivision_code = customer_obj.get('homeAddress').get('regionCode') +'-'+customer_obj.get('homeAddress').get('countryCode')
address_data['country_subdivision_code'] = customer_obj.get('homeAddress').get('regionCode')
address_data['postal_code'] = customer_obj.get('homeAddress').get('postalCode')
address_data['country_alpha_2_code'] = customer_obj.get('homeAddress').get('countryCode')
address_data['country'] = customer_obj.get('homeAddress').get('country')
address_data['country_alpha_2_code'] = customer_obj.get('homeAddress').get('countryCode')
address_data['line_1'] = cvent_contact_obj.get('homeAddress').get('address1')
address_data['line_2'] = cvent_contact_obj.get('homeAddress').get('address2')
address_data['line_3'] = cvent_contact_obj.get('homeAddress').get('address3')
address_data['city'] = cvent_contact_obj.get('homeAddress').get('city')
address_data['state_province'] = cvent_contact_obj.get('homeAddress').get('region')
country_subdivision_code = cvent_contact_obj.get('homeAddress').get('countryCode') +'-'+cvent_contact_obj.get('homeAddress').get('regionCode')
address_data['country_subdivision_code'] = country_subdivision_code
address_data['postal_code'] = cvent_contact_obj.get('homeAddress').get('postalCode')
address_data['country_alpha_2_code'] = cvent_contact_obj.get('homeAddress').get('countryCode')
address_data['country'] = cvent_contact_obj.get('homeAddress').get('country')
address_data['country_alpha_2_code'] = cvent_contact_obj.get('homeAddress').get('countryCode')
log.debug(address_data)
contact_data['address'] = address_data
person_data['contact'] = contact_data
membership = {}
membership['first_start_on'] = customer_obj.get('membership').get('joined') # A trigger should update 'first_start_on' if needed???
membership['start_on'] = customer_obj.get('membership').get('lastRenewal') # A trigger should update 'first_start_on' if needed???
membership['end_on'] = customer_obj.get('membership').get('expiration')
membership_person_data = {}
membership_person_data['first_start_on'] = cvent_contact_obj.get('membership').get('joined') # A trigger should update 'first_start_on' if needed???
membership_person_data['start_on'] = cvent_contact_obj.get('membership').get('lastRenewal') # A trigger should update 'first_start_on' if needed???
membership_person_data['end_on'] = cvent_contact_obj.get('membership').get('expiration')
log.debug(membership_person_data)
membership_type_id = customer_obj.get('type').get('id')
membership_type_name = customer_obj.get('type').get('name')
membership_type_cvent_id = cvent_contact_obj.get('type').get('id')
membership_type_cvent_name = cvent_contact_obj.get('type').get('name')
log.info(f'Found Cvent Membership Type Named: {membership_type_cvent_name}')
# 'id': '5EB898D8-C253-482C-A93A-0B6667C26E04', 'name': 'Al-Anon Member'
# 'id': 'A20358C5-0F6C-47AF-9843-BA9483A9D767', 'name': 'Al-Anon Non-Member'
# 'id': 'A01900AB-496A-48A1-9B04-C2874651227E', 'name': 'Member'
@@ -116,23 +145,101 @@ async def get_person(
# 'id': '71D07118-C24D-4B2E-888D-56AC1B941495', 'name': "IDAA 20's Guest Registration"
# 'id': 'DA17F721-9924-43E3-A31F-C567BA96DC64', 'name': 'IDAA Teen (13-19 years)'
if membership_type_name == 'Al-Anon Member' or membership_type_name == 'Al-Anon Members':
membership_person_data = {}
membership_person_type_data = {}
if membership_type_cvent_name == 'Al-Anon Member' or membership_type_cvent_name == 'Al-Anon Members':
# membership_person_data['membership_type_id'] = 6
membership_person_type_data['membership_type_id'] = 6
membership_person_type_data['product_id'] = 13
membership_person_data['level'] = 1
# membership_person_data['level'] = 1
membership_person_type_data['level'] = 1
elif membership_type_name == 'Annual Contribution' or membership_type_name == 'Annual Contributions': # Unsure... making affiliate
elif membership_type_cvent_name == 'Annual Contribution' or membership_type_cvent_name == 'Annual Contributions': # Unsure... making affiliate
# membership_person_data['membership_type_id'] = 8
membership_person_type_data['membership_type_id'] = 8
membership_person_type_data['product_id'] = 13
membership_person_data['level'] = 3
# membership_person_data['level'] = 3
membership_person_type_data['level'] = 3
elif membership_type_name == 'Doctoral Qualifying Member' or membership_type_name == 'Doctoral Qualifying Members':
membership_person_type_data['membership_type_id'] = 5
elif membership_type_cvent_name == 'Doctoral Qualifying Member' or membership_type_cvent_name == 'Doctoral Qualifying Members':
membership_person_data['membership_type_id'] = 5
# membership_person_type_data['membership_type_id'] = 5
membership_person_type_data['product_id'] = 4
membership_person_data['level'] = 1
# membership_person_data['level'] = 1
membership_person_type_data['level'] = 1
elif membership_type_name == 'Student Member' or membership_type_name == 'Student Members':
elif membership_type_cvent_name == 'Student Member' or membership_type_cvent_name == 'Student Members':
# membership_person_data['membership_type_id'] = 7
membership_person_type_data['membership_type_id'] = 7
membership_person_type_data['product_id'] = 14
membership_person_data['level'] = 1
membership_person_type_data['level'] = 1
# membership_person_data['level'] = 1
membership_person_type_data['level'] = 1
membership_person_type_data['first_start_on'] = datetime.datetime.strptime(cvent_contact_obj.get('membership').get('joined'), '%Y-%m-%d')
membership_person_type_data['start_on'] = datetime.datetime.strptime(cvent_contact_obj.get('membership').get('lastRenewal'), '%Y-%m-%d')
membership_person_type_data['end_on'] = datetime.datetime.strptime(cvent_contact_obj.get('membership').get('lastRenewal'), '%Y-%m-%d')
membership_person_type_data['last_end_on'] = datetime.datetime.strptime(cvent_contact_obj.get('membership').get('expiration'), '%Y-%m-%d')
current_datetime = datetime.datetime.now()
if membership_person_type_data['end_on'] >= current_datetime:
membership_person_type_data['lu_membership_type_status_id'] = 5 # 5 = active; expiration is > now
else:
membership_person_type_data['lu_membership_type_status_id'] = 7 # 7 = inactive; expiration is > now
membership_person_data['enable'] = True
membership_person_type_data['enable'] = True
membership_person_data['membership_person_type'] = membership_person_type_data
log.debug(membership_person_data)
if person_external_id := person_data.get('external_id') and 1==3:
log.info(f'Looking up person with External ID: {person_external_id}')
if result := get_person_rec_w_external_id(account_id=account_id, external_id=person_external_id):
log.debug(result)
person_id = result.get('person_id')
log.info(f'Person ID {person_id} found.')
person_obj = load_person_obj(person_id=person_id, inc_address=True, inc_contact=True, inc_membership_person=True, inc_membership_person_type=True)
log.debug(person_obj)
person_data['id'] = person_id
person_data['account_id'] = account_id
person_data['contact']['id'] = person_obj.contact.id
person_data['contact']['address']['id'] = person_obj.contact.address.id
log.debug(person_data)
membership_person_id = person_obj.membership_person.id
membership_person_data['id'] = person_obj.membership_person.id
membership_person_data['membership_person_type']['id'] = person_obj.membership_person.membership_person_type.id
if update_membership_person_obj(membership_person_id=membership_person_id, membership_person_dict_obj=membership_person_data):
log.info(f'Updated Membership Person ID: {membership_person_id}')
else:
log.info(f'Did not update Membership Person ID: {membership_person_id}')
person_data['membership_person'] = membership_person_data
return mk_resp(data=person_data, status_message='Updated and loaded person based on Cvent contact information', response=commons.response)
# try:
# person_obj = Person_Base(**person_data)
# log.debug(person_obj)
# except ValidationError as e:
# log.error(e.json())
# return False
update_person_kiss(person_id=person_id, person_dict_obj=person_obj)
person_obj = load_person_obj(person_id=person_id, inc_address=True, inc_contact=True)
log.debug(person_obj)
return mk_resp(data=person_obj, status_message='Updated and loaded person based on Cvent contact information', response=commons.response)
elif email := cvent_contact_obj.get('email'):
log.info(f'Looking up person with Email Address: {email}')
if result := get_person_rec_list(for_obj_type='account', for_obj_id=account_id, email=email):
log.debug(result[0])
person_id = result[0].get('person_id')
log.info(f'Person ID {person_id} found.')
else:
log.info(f'External ID and Email Address were not found.')
return mk_resp(data=person_data, status_message='Got Cvent contact information', response=commons.response)