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

Unable to use one page for multiple DEV models that are partially in sync
Hello, Our model builders are normally making changes to all structural changes in a DEV model. Because we are now in a Prod freeze and we don't want to move structural changes to PROD until end of year, we created a copy of the DEV model (standard mode) - 'DEV2' model, where we are only making changes to urgent issues…
Anaplan for Microsoft 365
Using the excel addin for Microsoft 365 when I try to add a card to import data I get the message 'Cannot insert data from a custom view'. I am not using a custom view. Any ideas what I am missing here?
Data Split between Many to Many Relationship Mappings
Hello Anaplan Community, I am encountering an issue with one of my development work and would be greatly appreciated your assistance. There are 2 different modules involved in my model: 1) INP01 Percentage Input by Product (Applies To: P3 List) 2) REP01 Percentage Split by Department (Applies To: Departments List) The…