Skip to content

Instantly share code, notes, and snippets.

Embed
What would you like to do?
import requests, json, math, datetime
# global variables
api_base_url = 'https://fivetran.namely.com/api/v1'
response_per_page = 20
def handler(request):
# get the credentials needed securely from Fivetran function // no storing credentials in code
request_json = request.get_json(silent=True)
bearer_token = request_json['secrets']['bearer_token']
headers = {
'Authorization': 'Bearer ' + bearer_token
}
################## Profiles Endpoint ##################
payload = {}
profiles_endpoint = 'profiles'
# determine the number of pages needed for the specified response_per_page
request_url_for_meta = generate_request_url(api_base_url, profiles_endpoint)
meta_response = requests.request("GET", request_url_for_meta, headers=headers, data=payload).text.encode('utf8')
parsed_meta_response = json.loads(meta_response)
total_element_count = int(parsed_meta_response['meta']['total_count'])
number_of_pages_needed = math.ceil(total_element_count / response_per_page)
# paginate through number of pages and combine responses
# i is the page number
profiles_all = json.loads('[]')
for i in range(1, number_of_pages_needed + 1, 1):
print('going through page ' + str(i))
params = f"page={str(i)}&per_page={str(response_per_page)}"
request_url_for_profiles_i = generate_request_url(api_base_url, profiles_endpoint, params)
profile_response_i = requests.request("GET", request_url_for_profiles_i, headers=headers,
data=payload).text.encode(
'utf8')
parsed_profile_response_i = json.loads(profile_response_i)
profiles_i = parsed_profile_response_i['profiles']
profiles_all = profiles_all + profiles_i
# loop through each profile and extract only the necessary fields
profiles_table = []
for profile in profiles_all:
profile_extract = {
'id': profile['id'],
'user_status': profile['user_status'],
'start_date': profile['start_date'],
'departure_date': profile['departure_date'],
'gender': profile['gender'],
'employee_type': profile['employee_type']['title'],
'employee_id': profile['employee_id'],
'division': profile['division']
}
profiles_table.append(profile_extract)
################## Fivetran Response JSON ##################
now = datetime.datetime.now().isoformat()
response = {}
response['schema'] = {
'profile': {'primary_key': ['id']}
}
response['insert'] = {}
response['insert']['profile'] = profiles_table
# TODO hit this endpoint next
# response['insert']['group_member'] = group_member_table
response['state'] = {}
response['state']['cursor'] = now
response['hasMore'] = False
return json.dumps(response, indent=2), 200, {"Content-Type": "application/json"}
# a general request url generator that includes the base, the endpoint, and optional params and filters
def generate_request_url(base_url, endpoint, params=None, filters=None):
# example parameters
# base_url = "https://fivetran.namely.com/api/v1"
# endpoint = "profiles"
# params (optional) = "page=1&per_page=20"
# filters (optional) = "[user_status]=active"
combined = base_url + '/' + endpoint + '.json'
if params:
combined = combined + '?' + params
if filters:
combined = combined + '&filter' + filters
return combined
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment