REST API : Recursive function with Python

Hi,

I'm currently working on REST API. I would like to create a file with last username connected to each model.

Based on something created by Karan Kochhar, I currently have extracted all model information (with the user id of the last modification)

Problem is I can't extract all users information and merge both array.

I assume I need to create a loop to get the name of the user for each model.

I tried this code but it's not working. Can you help to fix this issue ?

Thanks

Regards,

Yohan

models_response = anaplan.getModel(token)

for a in models_response['models']:

if a['activeState'] != 'ARCHIVED':

user_info_response = anaplan.getUsersInformation(a['lastModifiedByUserGuid'],token)

models_df =pd.DataFrame(models_response['models'])

output_df = pd.merge(models_df,user_info_response[['id','email','lastname','firstname']],left_on='lastModifiedByUserGuid', right_on='id')

------

def getModel(vToken):

model_url = 'https://api.anaplan.com/2/0/models/?modelDetails=True'

model_header = {'Authorization':'AnaplanAuthToken ' + vToken}

r = requests.get(model_url, headers=model_header)

try:

r.raise_for_status()

except requests.exceptions.HTTPError as e:

print("Error: " + str(e))

exit()

return r.json()

def getUsersInformation(UserID,vToken):

getUsersInformation_url = 'https://api.anaplan.com/2/0/users/'+UserID

header = {'Authorization':'AnaplanAuthToken ' + vToken,'Accept':'application/json'}

r = requests.get(getUsersInformation_url, headers= header)

try:

r.raise_for_status()

except requests.exceptions.HTTPError as e:

print("Error: " + str(e))

exit()

return r.json()

Accepted answers

karanAMP

@yrignac

When it comes to API call , I cannot completely test this as I am not a tenant admin(I get 404 error as I don't have access to every user detail) However i have tested this with my info only and it works .

With regard to your python code , i think there are few issues

for a in models_response['models']:
if a['activeState'] != 'ARCHIVED':
user_info_response = anaplan.getUsersInformation(a['lastModifiedByUserGuid'],token)
models_df =pd.DataFrame(models_response['models'])
output_df = pd.merge(models_df,user_info_response[['id','email','lastname','firstname']],left_on='lastModifiedByUserGuid', right_on='id')
you are recursively "over writing" user_info_response by calling the getUsers Information function in for loop . What you should do is append the responses in python object ( list )

user_info_response = []

for a in models_response['models'] :
   if a['activeState'] != 'ARCHIVED':
        user_info_response.append(getUsersInformation(a['lastModifiedByUserGuid'],token)['user'])

here I have defined an empty list user_info_response and then i am "appending" to the empty list . Also note ['user'] in the end this will give you only 'user' json tags. This should give you something like a list of dictionaries

for a in models_response['models']:

if a['activeState'] != 'ARCHIVED':

user_info_response = anaplan.getUsersInformation(a['lastModifiedByUserGuid'],token)

models_df =pd.DataFrame(models_response['models'])

output_df = pd.merge(models_df,user_info_response[['id','email','lastname','firstname']],left_on='lastModifiedByUserGuid', right_on='id')

in above highlighted code . model_df and output_df should never be in for loop . Instead do it outside for loop

user_info_df = pd.DataFrame(user_info_response)
models_df =pd.DataFrame(models_response['models'])
output_df = pd.merge(models_df,user_info_df,left_on='lastModifiedByUserGuid', right_on='id',how='left')

here I have converted user_info_response list to user_info_df

model_df have model data

output.df is a merge of model_df and user_info_df . Please note how = left clause

complete code

import pandas as pd
import requests
import json
from base64 import b64encode

###############################################
username = "your username"
password = "your password"

###############################################

def getToken():
        
    header_string = { 'Authorization':'Basic ' + b64encode((username + ":" + password).encode('utf-8')).decode('utf-8') }
    anaplan_url='https://auth.anaplan.com/token/authenticate'

    print('getting Token')

    r=requests.post(anaplan_url, headers=header_string)

    try:
        r.raise_for_status()
    except requests.exceptions.HTTPError as e:
        print("Error: " + str(e))
        exit()

    resp_dict = json.loads(r.text)
    return resp_dict['tokenInfo']['tokenValue']

def getWorkspaces(t):

    workspace_url = 'https://api.anaplan.com/2/0/workspaces/?tenantDetails=true'
    header = {'Authorization':'AnaplanAuthToken ' + t}
    r = requests.get(workspace_url,headers=header)

    try:
        r.raise_for_status()
    except requests.exceptions.HTTPError as e:
        print("Error: " + str(e))
        exit()

    return r.json()

def getModel(t):
    
    model_url = 'https://api.anaplan.com/2/0/models/?modelDetails=True'
    model_header = {'Authorization':'AnaplanAuthToken ' + t}
    r = requests.get(model_url, headers=model_header)
    
    try:
        r.raise_for_status()
    except requests.exceptions.HTTPError as e:
        print("Error: " + str(e))
        exit()
    
    return r.json()

def getUsersInformation(UserID,vToken):
    getUsersInformation_url = 'https://api.anaplan.com/2/0/users/'+UserID
    header = {'Authorization':'AnaplanAuthToken ' + vToken,'Accept':'application/json'}
    r = requests.get(getUsersInformation_url, headers= header)

    try:
        r.raise_for_status()
    except requests.exceptions.HTTPError as e:
        print("Error: " + str(e))
        exit()
    return r.json()

################# main ##########################

token = getToken()
print(token)

workspaces_response = getWorkspaces(token)

#Saving workspace data in dataframe
workspace_df = pd.DataFrame(workspaces_response['workspaces'])
print(workspace_df)

models_response = getModel(token)

user_info_response = []
for a in models_response['models'] :
   if a['activeState'] != 'ARCHIVED':
        user_info_response.append(getUsersInformation(a['lastModifiedByUserGuid'],token)['user'])
        
user_info_df = pd.DataFrame(user_info_response)
models_df =pd.DataFrame(models_response['models'])
output_df = pd.merge(models_df,user_info_df,left_on='lastModifiedByUserGuid', right_on='id',how='left')

All comments

yrignac

Nobody has ever exeperience that ?

Thank you for your help

karanAMP

@yrignac

With regard to your python code , i think there are few issues

for a in models_response['models']:
if a['activeState'] != 'ARCHIVED':
user_info_response = anaplan.getUsersInformation(a['lastModifiedByUserGuid'],token)
models_df =pd.DataFrame(models_response['models'])
output_df = pd.merge(models_df,user_info_response[['id','email','lastname','firstname']],left_on='lastModifiedByUserGuid', right_on='id')
you are recursively "over writing" user_info_response by calling the getUsers Information function in for loop . What you should do is append the responses in python object ( list )

user_info_response = []

for a in models_response['models'] :
   if a['activeState'] != 'ARCHIVED':
        user_info_response.append(getUsersInformation(a['lastModifiedByUserGuid'],token)['user'])

for a in models_response['models']:

if a['activeState'] != 'ARCHIVED':

user_info_response = anaplan.getUsersInformation(a['lastModifiedByUserGuid'],token)

models_df =pd.DataFrame(models_response['models'])

output_df = pd.merge(models_df,user_info_response[['id','email','lastname','firstname']],left_on='lastModifiedByUserGuid', right_on='id')

in above highlighted code . model_df and output_df should never be in for loop . Instead do it outside for loop

user_info_df = pd.DataFrame(user_info_response)
models_df =pd.DataFrame(models_response['models'])
output_df = pd.merge(models_df,user_info_df,left_on='lastModifiedByUserGuid', right_on='id',how='left')

here I have converted user_info_response list to user_info_df

model_df have model data

output.df is a merge of model_df and user_info_df . Please note how = left clause

complete code

import pandas as pd
import requests
import json
from base64 import b64encode

###############################################
username = "your username"
password = "your password"

###############################################

def getToken():
        
    header_string = { 'Authorization':'Basic ' + b64encode((username + ":" + password).encode('utf-8')).decode('utf-8') }
    anaplan_url='https://auth.anaplan.com/token/authenticate'

    print('getting Token')

    r=requests.post(anaplan_url, headers=header_string)

    try:
        r.raise_for_status()
    except requests.exceptions.HTTPError as e:
        print("Error: " + str(e))
        exit()

    resp_dict = json.loads(r.text)
    return resp_dict['tokenInfo']['tokenValue']

def getWorkspaces(t):

    workspace_url = 'https://api.anaplan.com/2/0/workspaces/?tenantDetails=true'
    header = {'Authorization':'AnaplanAuthToken ' + t}
    r = requests.get(workspace_url,headers=header)

    try:
        r.raise_for_status()
    except requests.exceptions.HTTPError as e:
        print("Error: " + str(e))
        exit()

    return r.json()

def getModel(t):
    
    model_url = 'https://api.anaplan.com/2/0/models/?modelDetails=True'
    model_header = {'Authorization':'AnaplanAuthToken ' + t}
    r = requests.get(model_url, headers=model_header)
    
    try:
        r.raise_for_status()
    except requests.exceptions.HTTPError as e:
        print("Error: " + str(e))
        exit()
    
    return r.json()

def getUsersInformation(UserID,vToken):
    getUsersInformation_url = 'https://api.anaplan.com/2/0/users/'+UserID
    header = {'Authorization':'AnaplanAuthToken ' + vToken,'Accept':'application/json'}
    r = requests.get(getUsersInformation_url, headers= header)

    try:
        r.raise_for_status()
    except requests.exceptions.HTTPError as e:
        print("Error: " + str(e))
        exit()
    return r.json()

################# main ##########################

token = getToken()
print(token)

workspaces_response = getWorkspaces(token)

#Saving workspace data in dataframe
workspace_df = pd.DataFrame(workspaces_response['workspaces'])
print(workspace_df)

models_response = getModel(token)

user_info_response = []
for a in models_response['models'] :
   if a['activeState'] != 'ARCHIVED':
        user_info_response.append(getUsersInformation(a['lastModifiedByUserGuid'],token)['user'])
        
user_info_df = pd.DataFrame(user_info_response)
models_df =pd.DataFrame(models_response['models'])
output_df = pd.merge(models_df,user_info_df,left_on='lastModifiedByUserGuid', right_on='id',how='left')

yrignac

Well thank you very much Karan.

I had to split the for loop to avoid duplicate member as I have several models with the same last modified id user.

I still have one specific for one model who the user has been removed.

Id exists within the model list but when you get user information with the id we got a error.

But I'll find a solution by myself.

Thank you very much

Regards,

Yohan

Quick Links

Can you help?

Unanswered questions

Predefaulting A Feild in ANAPLAN
Hello Anaplan Community, I have a special requirement from a customer and would appreciate your input. Let me explain the situation: We have a hierarchy with three levels: Level 1, Level 2, and Level 3. Screen 1: Creation of Level 1 Items On this screen, the user selects a team. Based on their selection, an action runs to…
Looking for Anaplan End-Users to take a short survey!
We are looking for Anaplan end-users to provide feedback on their experiences with the Excel add-in. Interested individuals will respond to this 5-minute survey to help us understand personal needs and behavior when using the add-in. The feedback provided by survey takers is essential to the roadmap of Anaplan's products.…
CERTIFICATION BADGES - ANNOUNCEMENT
Anaplan Champions! The Community team just posted this announcement that certification badges may not be showing up on your profile probably until next year. Rest assured, you will get credit once you complete and pass the exams. https://community.anaplan.com/t5/Blog/Badges-Back-Soon/ba-p/123385