Initiating Integration API Runs with Python

Overview

The goal of this tutorial is to provide a working example of how 'runs' of the LeanIX integration API can be initiated from a 3rd party system using Python.

❗️

Command Line Interface (CLI) Version

This tutorial involves modifying/assigning variables in a Python script. For a version of this that can be executed in the CLI without directly modifying code, please visit our Public Scripts repository.

📘

Processor Details

Detailed documentation about configuring Integration API connectors are not covered in this tutorial. Comprehensive documentation of the Integration API can be found here.

Prerequisites

  • The Integration API enabled in your LeanIX workspace (for this, contact your Customer Success Manager).
  • An outbound and/or inbound processor configured in your workspace to test out, depending on your requirements.
  • Python 3 installed on your system.
  • Python Libraries installed: json, requests, time.

Initial Script Setup

The script initiateOutboundRun.py below provides a template which can be customized to initiate either inbound and outbound run to/from the Integration API.

🚧

Outbound vs Inbound

This tutorial focuses initially on initiating an outbound run. Please continue to the bottom of the page for instructions on adapting the script for inbound purposes.

  • Download the script below open it in a text editor of your choosing (we use Visual Studio Code).

  • Update the request_url and auth_url variables on lines 5 and 6 with your workspace domain. You can find this by looking at your LeanIX workspace URL.

  • Generate an API Token within your workspace (Admin > API Tokens) and assign in to the api_token variable (line 9)

import json 
import requests 
import time

auth_url = 'https://<your domain>.leanix.net/services/mtm/v1/oauth2/token' 
request_url = 'https://<your domain>.leanix.net/services/integration-api/v1/' 


api_token = '<your api token>'

response = requests.post(auth_url, auth=('apitoken', api_token),
                         data={'grant_type': 'client_credentials'})

response.raise_for_status() 
access_token = response.json()['access_token']
auth_header = 'Bearer ' + access_token


header = {'Authorization': auth_header, 'Content-Type': 'application/json'}


def call_post(endpoint, data=False):
    response = requests.post(url=request_url + endpoint, headers=header, data=data)
    response.raise_for_status()
    return response


def call_get(endpoint):
    response = requests.get(url=request_url + endpoint, headers=header)
    response.raise_for_status()
    return response

def create_run(run_config):
    result = call_post("synchronizationRuns", json.dumps(run_config))
    return json.loads(result.text)['id']

def start_run(run_id):
    start_run_endpoint = 'synchronizationRuns/%s/start' % (run_id)
    result = call_post(start_run_endpoint)
    return result.status_code

def check_run_status(run_id, status_response=None):
    print('checking status')
    status_endpoint = 'synchronizationRuns/%s/status' % (run_id)
    status_response = call_get(status_endpoint)
    status_response = json.loads(status_response.text)['status']
    print(status_response)
    if status_response != 'FINISHED':
        time.sleep(5)
        return check_run_status(run_id, status_response)
    else:
        return True
     
def fetch_results(run_id):
    results_endpoint = 'synchronizationRuns/%s/results' % (run_id)
    results_response = call_get(results_endpoint)
    return json.loads(results_response.text)


def handle_run(ldif_data):
    run_id = create_run(ldif_data)
    if start_run(run_id) == 200:
        if check_run_status(run_id) == True:
            if ldif_data['processingDirection'] == "outbound":
                return fetch_results(run_id)
            elif ldif_data['processingDirection'] == "inbound":
                return f"inbound run: {run_id} finished successfully"



connectorType = ""
connectorId = ""
connectorVersion = ""
lxVersion = "1.0.0"
processingDirection = ""
#inbound or outbound
processingMode = ""
#partial or full 
description = ""
contents = []
### contents array will be left empty for outbound runs

run_config = {
      "connectorType": connectorType,
      "connectorId": connectorId,
      "connectorVersion": connectorVersion,
      "lxVersion": lxVersion,
      "description": description,
      "processingDirection": processingDirection,
      "processingMode": processingMode,
      "content": contents
    }


run_results = handle_run(run_config)

with open('leanixOutboundData' + connectorVersion + '.json', 'w') as outfile:
    json.dump(run_results, outfile, ensure_ascii=False, indent=4)

Adjust Connector Variables

The variables on lines 69 - 74 are used to identify the proper Workspace and Connector to be triggered by the script.

  • Variables connectorType, connectorId, connectorVersion, processingDirection, and processingMode should all be adapted to match the Connector you want to trigger and can be found within the Integration API user interface (Admin > Integration API).

  • lxWorkspace should be set to your workspace ID, which can be found in Admin > API Tokens.

📘

LDIF Elements

Additional information about the LDIF configuration elements can be found here.

With these variables in place, you can run the script and it will call the Integration API, initiating a 'run' of the processor indicated per your configured variables. The data retrieved from the run will be saved in the same directory as your copy of initiateOutboundRun.py in a file called leanixOutboundData.json

Adapt for Inbound

If you would like to submit an LDIF to an inbound connector, this is possible with the following steps:

  • Adjust your configuration variables to point the script at an inbound processor

  • Add the key "contents": [] to the run_config object on line 87

  • Submit your LDIF formatted data within the contents array (see example below)

contents:[
  {
      "type": "ITComponent",
      "id": "b6992b1d-4e4d",
      "data": {
        "name": "Gatsby.j",
        "description": "Gatsby is a free and open source framework based on React that helps developers build websites and apps.",
        "category": "sample_software",
        "provider": "gcp",
        "applicationId": "28db27b1-fc55-4e44"
      }
    },
    {
      "type": "ITComponent",
      "id": "cd4fab6c-4336",
      "data": {
        "name": "Contentful",
        "description": "Beyond headless CMS, Contentful is an API-first content management infrastructure to create, manage and distribute content to any platform or device.",
        "category": "cloud_service",
        "provider": "gcp",
        "applicationId": "28db27b1-fc55-4e44"
      }
    }
]

🚧

You will likely not want to 'hard code' your LDIF in the run_config in any substantial implementation, as this would be inflexible and not scalable.

If you are considering implementing an integration with LeanIX Integration API, please don't hesitate to reach out to your Customer Success Manager to arrange a meeting with a member of our Engineering team for a consultation.