Python Anaplan Connector

riyaserenajo
riyaserenajo
in Integrations and extensibility

Im trying to learn about the PYTHON Anaplan Connector, can someone please share the correct documentation starting from scratch and best practices using PYTHON Connector?

I need to understand about File upload, File Export PYTHON SCRIPTS how we can implement and what are the Prerequisites.

Tagged:

Answers

  • frank58richard
    frank58richard

    Integrating Anaplan with Python enables efficient automation of tasks such as file uploads and exports. Here's a structured approach to get you started:

    1. Understanding the Anaplan API: Anaplan provides a RESTful API that allows programmatic access to its platform. Familiarize yourself with the Anaplan API 2.0 documentation to understand the available endpoints and their functionalities.
    2. Setting Up the Python Environment: Ensure you have Python installed on your system. It's recommended to use a virtual environment to manage dependencies. You can install necessary libraries using pip:
    3. pip install requests
    4. Authentication: Anaplan supports multiple authentication methods, including Basic Authentication and Certificate-based Authentication. Choose the method that aligns with your organization's security protocols. For instance, to use Basic Authentication:
      import requests

    url = 'https://api.anaplan.com/2/0/authentication/token'
    headers = {
    'Authorization': 'Basic YOUR_BASE64_ENCODED_CREDENTIALS'
    }
    response = requests.post(url, headers=headers)
    if response.status_code == 200:
    auth_token = response.json()['token']
    else:
    print(f"Authentication failed: {response.status_code}")
    4. File Upload:

    To upload a file to Anaplan, you'll need to create a file object and then initiate an import action. Here's a simplified example:

    import requests

    first step

    file_path = 'path_to_your_file.csv'
    url = 'https://api.anaplan.com/2/0/workspaces/{workspace_id}/models/{model_id}/files'
    headers = {
    'Authorization': 'Bearer YOUR_AUTH_TOKEN',
    'Content-Type': 'application/json'
    }
    files = {'file': open(file_path, 'rb')}
    response = requests.post(url, headers=headers, files=files)
    if response.status_code == 201:
    file_id = response.json()['id']
    print(f"File uploaded successfully with ID: {file_id}")
    else:
    print(f"File upload failed: {response.status_code}")
    Replace {workspace_id} and {model_id} with your specific Anaplan workspace and model IDs.

    1. Initiating an Import Action:

    After uploading the file, you can initiate an import action to process the data:

    python
    Copy
    import requests
    import json

    second step

    import_action_id = 'your_import_action_id'
    url = f'https://api.anaplan.com/2/0/workspaces/{workspace_id}/models/{model_id}/imports/{import_action_id}/tasks'
    headers = {
    'Authorization': 'Bearer YOUR_AUTH_TOKEN',
    'Content-Type': 'application/json'
    }
    data = {
    "localeName": "en_US",
    "fileIds": [file_id]
    }
    response = requests.post(url, headers=headers, data=json.dumps(data))
    if response.status_code == 201:
    task_id = response.json()['id']
    print(f"Import action initiated successfully with Task ID: {task_id}")
    else:
    print(f"Import action initiation failed: {response.status_code}")
    6. Monitoring Import Status:

    To monitor the status of the import task:

    python
    Copy
    import requests

    Third step

    url = f'https://api.anaplan.com/2/0/workspaces/{workspace_id}/models/{model_id}/tasks/{task_id}'
    headers = {
    'Authorization': 'Bearer YOUR_AUTH_TOKEN'
    }
    response = requests.get(url, headers=headers)
    if response.status_code == 200:
    task_status = response.json()['status']
    print(f"Import task status: {task_status}")
    else:
    print(f"Failed to retrieve task status: {response.status_code}")
    7. File Export:

    To export data from Anaplan:

    python
    Copy
    import requests

    Fourth step

    export_action_id = 'your_export_action_id'
    url = f'https://api.anaplan.com/2/0/workspaces/{workspace_id}/models/{model_id}/exports/{export_action_id}/tasks'
    headers = {
    'Authorization': 'Bearer YOUR_AUTH_TOKEN',
    'Content-Type': 'application/json'
    }
    data = {
    "localeName": "en_US"
    }
    response = requests.post(url, headers=headers, data=json.dumps(data))
    if response.status_code == 201:
    task_id = response.json()['id']
    print(f"Export action initiated successfully with Task ID: {task_id}")
    else:
    print(f"Export action initiation failed: {response.status_code}")
    8. Downloading Exported File:

    After initiating the export, download the file:

    python
    Copy
    import requests

    fifth step

    url = f'https://api.anaplan.com/2/0/workspaces/{workspace_id}/models/{model_id}/exports/{export_action_id}/files/{file_id}'
    headers = {
    'Authorization': 'Bearer YOUR_AUTH_TOKEN',
    'Accept': 'application/octet-stream'
    }
    response = requests.get(url, headers=headers)
    if response.status_code == 200:
    with open('exported_file.csv', 'wb') as f:
    f.write(response.content)
    print("File downloaded successfully.")
    else:
    print(f"File download failed: {response.status_code}")
    Prerequisites:

    Hope that helps!

    Best Regard,

    Frank

    HMFusa

  • ChrisStauffer
    ChrisStauffer

    you can also download the Official Anaplan Postman Collection to help get enabled.

    https://www.postman.com/apiplan/official-anaplan-collection

Categories