Showing results for 
Search instead for 
Did you mean: 
Community Manager
Community Manager


Working with the Incorta APIs can be a handy and extensible way to manage your workflow scheduling and ordering. While in this article, we will touch on a script that will kick off multiple schemas, be aware there is also an endpoint to retrieve schema status.

Note: This capability is only available in the cloud, version 2022.11 and later.

What you need to know before reading this article

Administrative steps:

  1. Ensure the Public APIs are enabled for your cluster
  2. Ensure that users have permission to generate an API key 

There are a few things you'll need to ensure that your script can run successfully:

  1. Incorta URL - the core URL ending in ".com". remove any sub-directories.
  2. Tenant - Can be found as part of your login page for Incorta 
  3. Username - Also can be found as part of your login
  4. API Key - see above administrative steps
  5. Schemas - names of schemas that will be loaded via this script (case-sensitive).

Let's Go

All that needs to be done is to plug in the above five variables into this script and run! Ensure that your cluster is connected when running this script. Notice that the script's first half is building out functions to format the request.  The changes should all happen in the last ~20 lines of python. 




import json
import requests
import urllib3
def request_tokens(hostname, tenant, username, api_key):
    import urllib3

    tokens_url = hostname + '/incorta/api/v1/tokens'
    headers = {'Content-type': 'application/json'}
    request_payload = construct_authentication_request(tenant, username, api_key)

    urllib3.disable_warnings() # Prevent InsecureRequestWarning

    result =, data=request_payload, headers=headers, verify=False)

    if result.status_code != 201:
        print("Error while requesting tokens: ")
        print("Response code from server is: " + str(result.status_code))
        print("Response body is: " + result.text)
        return "Error"
        tokens_dictionary = json.loads(result.text)
        return tokens_dictionary['accessToken']

def construct_authentication_request(tenant, username, api_key):
    import hashlib
    import time
    import calendar

    timestamp = calendar.timegm(time.gmtime()) * 1000  # convert to milliseconds
    print("timestamp: " + str(timestamp))
    request = api_key + str(timestamp)
    request_hash = hashlib.sha256(request.encode('utf-8')).hexdigest()

    request_payload = {
        'tenantName': tenant,
        'loginName': username,
        'requestHash': request_hash,
        'timestamp': timestamp
    return json.dumps(request_payload)

# Begin Schema Load Request
api_key = {INSERT API KEY HERE IN ""}
incorta_url = {INSERT URL HERE IN ""}
access_token = request_tokens(incorta_url, tenant, username, api_key)
headers = {"Authorization": "Bearer "+ access_token, "Accept":"application/json", "Content-type": "application/json"}
schema_request_payload = {
    "schemaNames":["{SCHEMA NAME}","{SCHEMA NAME}"],
    "loadSource": "FULL",
    "synchronous": True,
    "requestedTimeoutInMilliseconds": 0
schema ="/incorta/api/v1/schema/load", json=schema_request_payload, headers=headers)




Related Material

Best Practices Index
Best Practices

Just here to browse knowledge? This might help!

Version history
Last update:
‎08-22-2023 12:44 PM
Updated by: