Anaplan API 2.0 Python Library - [Updated March 2022]

After a significant amount of refactoring, I'm very pleased to announce the latest release of my Python package for the Anaplan Bulk API. This package simplifies interacting with the Anaplan Auth and Bulk APIs. It supports Basic and Certificate authentication, certificates can be used directly or within a Java Keystore file. It supports more robust logging and contains more detailed documentation for its use. This package is now available on GitHub and for download using pip.

Note: The package used to handle Java Keystore files appears to be an inactive project and contains a bug that may cause issues during installation. Details for working around these issues can be found in the Github and PyPi pages for the Anaplan API package.

Note: While all of these scripts have been tested and found to be fully functional, due to the vast amount of potential use cases, Anaplan does not explicitly support custom scripts built by our customers. This article is for information only and does not suggest any future product direction. This library is a work in progress and will be updated with new features once they have been tested.

Getting Started

The attached Python library serves as a wrapper for interacting with the Anaplan API. This article will explain how you can use the library to automate many of the requests that are available in our Apiary, which can be found at https://anaplanbulkapi20.docs.apiary.io/#.

This article assumes you have the requests, cryptography, and pyjks modules installed, as well as Python 3.7 or later. Please make sure you are installing these modules with Python 3, and not for an older version of Python. For more information on these modules, please see their respective websites:

Gathering the Necessary Information

In order to use this library, the following information is required:

  • Anaplan model ID
  • Anaplan workspace ID
  • Anaplan action ID
  • CA certificate key-pair (private key and public certificate), or username and password

There are two ways to obtain the model and workspace IDs:

  1. While the model is open, go Help>About: Help Menu.png
  2. Select the workspace and model IDs from the URL: URL.png

Authentication

Every API request is required to supply valid authentication. There are two (2) ways to authenticate:

  1. Certificate Authentication
  2. Basic Authentication

For full details about CA certificates, please refer to our Anapedia article.

Basic authentication uses your Anaplan username and password.

To create a connection with this library, define the authentication type and details, and the Anaplan workspace and model IDs:

Certificate Files:

 

auth = anaplan.generate_authorization(auth_type='Certificate', cert='path/to/cert.pem', private_key='path/to/key.pem')
conn = AnaplanConnection(authorization=auth, workspace_id='<Workspace ID>', model_id='<Model ID>')
auth = anaplan.generate_authorization(auth_type='Basic', email='user@domain.com', password='password')
conn = AnaplanConnection(authorization=auth, workspace_id='<Workspace ID>', model_id='<Model ID>')

 

 

 Java Keystore:

 

 

from anaplan_api.KeystoreManager import KeystoreManager

keys = KeystoreManager(path='path/to/keystore.jks', passphrase='<Keystore password>', alias='<Key Alias>', key_pass='<Private key password>')
auth = anaplan.generate_authorization(auth_type='Certificate', cert=keys.get_cert(), private_key=keys.get_key())
conn = AnaplanConnection(authorization=auth, workspace_id='<Workspace ID>', model_id='<Model ID>')

 

Getting Anaplan Resource Information

You can use this library to fetch the list of resources within the specified Anaplan model: files, imports, exports, actions, processes:

Example:

 

files_list = anaplan.get_list(conn=conn, resource="files")

 

This returns a data object which contains a list of the specified resource. You can simply treat it like a dictionary, passing in the resource name and the object will return the corresponding ID.

 

print(files_list["Users.csv"])

 

Uploads

You can upload a file of any size, in chunks between 1-50mb. This upload can be both local files, or strings of data held in memory. The package will continue reading from the file/buffer in specified increments until all data is uploaded to your Anaplan model.
Flat file: 

 

 

anaplan.file_upload(conn=conn, file_id=files_list["Users.csv"], chunk_size=2, data='/path/to/file.csv')

 

"Streamed" file:

 

with open('/path/to/file.csv', 'r') as current_csv:
    data = current_csv.read()
    anaplan.file_upload(conn=conn, file_id=files_list["Users.csv"], chunk_size=2, data=data)

 

In the above example, both use local files but demonstrate how you can pass the data in as a string or simply provide the path to the local file. Once the upload is complete, you can move on to running your import action(s).

Executing Actions

You can run any Anaplan action with this script and define a number of times to retry the request if there's a problem. In order to execute an Anaplan action, the ID is required. To execute, all that is required is the following:

 

 

results = anaplan.execute_action(conn=conn, action_id="118000000007", retry_count=3)

#For Anaplan Processes
for item in results:
    print(item)

#For other Anaplan actions
print(results)

 

Example output:

Screenshot 2022-02-24 at 16.56.10.png

 

Once the specified action is complete, an object with the overall task results, task details, the file download (if applicable), and error dump (if applicable) is returned. Error dumps are only available for import actions, and are returned in a Pandas DataFrame. If the action is an export, the file is automatically downloaded and available as a string.

Downloading a File

There is no longer a dedicated need to download files, as export actions automatically pull the specified file. However, if you wish to download a file from Anaplan, you may do so.

 

 

file = anaplan.get_file(conn=conn, file_id="<File ID>")

 

 

Get Available Workspaces and Models

API 2.0 introduced a new means of fetching the workspaces and models available to a given user. You can use this library to build a key-value dictionary (as above) for these resources.

 

 

from anaplan_api.User import User
from anaplan_api.Workspace import Workspace
from anaplan_api.Model import Model

current_user = User(conn=conn)
current_user.get_current_user()

workspace = Workspace(conn=conn, user_id=current_user.get_id())
workspaces = workspace.get_workspaces()

print(workspaces)

model = Model(conn=conn, user_id=current_user.get_id())
models = model.get_models()

print(models)

 

 

This example will request details for current user, then fetch the workspaces and models available to that user.

Example Code

Screenshot 2022-02-24 at 16.49.25.png

 

Got feedback on this content? Let us know in the comments below.

The content in this article has not been evaluated for all Anaplan implementations and may not be recommended for your specific situation.
Please consult your internal administrators prior to applying any of the ideas or steps in this article.
Comments

Thanks for the library! For the get_file function, we are struggling when the export contains a comma. The function uses the results of the get .text to parse this out, but it is done by comma separated values. Is there a way to add a text qualifier? Thanks

@CommunityMember81209 

 

Sorry, I can't reproduce this. Whether the comma occurs in a text field or number, the payload of my file is correct (both CSV and TXT). A text delimiter is added to text fields that contain commas, and the commas is dropped from numeric fields (commas as decimal point are converted to period). 

Hello, We have found an issue in the get_file function. The URL needs to be inside the while loop that grabs each chunk and the file needs to be appended with each chunk of data. New code: 

#===========================================================================
# This function downloads a file from Anaplan to a string.
#===========================================================================
def get_file(conn, fileId):
'''
:param conn: AnaplanConnection object which contains authorization string, workspace ID, and model ID
:param fileId: ID of the Anaplan file to download
'''

chunk = 0
details = get_file_details(conn, fileId)
chunk_count = details[0]
file_name = details[1]

authorization = conn.authorization
workspaceGuid = conn.workspaceGuid
modelGuid = conn.modelGuid

get_header = {
"Authorization": authorization,
}
file = ""
print("Fetching file " + fileId + "...")

while int(chunk) < int(chunk_count):
url = __base_url__ + "/" + workspaceGuid + "/models/" + modelGuid + "/files/" + fileId + "/chunks/" + str(chunk)
try:
file_contents = requests.get(url, headers=get_header)
file_contents.raise_for_status()
except HTTPError as e:
raise HTTPError(e)
if file_contents.ok:
file = file + file_contents.text
else:
return "There was a problem fetching the file: " + file_name
break
chunk = str(int(chunk) + 1)

if int(chunk) == int(chunk_count):
print("File download complete!")

return file

@CommunityMember81209  I am also struggling with comma separated values.  I can usually fix it using parsing code in python, but it's not very efficient, and also very troublesome. 

 

What else do I need to do for file_contents.text to not include commas in the content?

Version history
Last update:
‎03-22-2022 09:48 PM
Updated by:
About the Author
  • 6 year veteran of Anaplan, working in Professional Services, formerly Customer Care. Subject matter expert in data integrations, with emphasis on API.

Contributors