Scheduling Schema Loads via the Incorta Public API...

JoeM · ‎01-12-2023

Introduction

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:

Ensure the Public APIs are enabled for your cluster
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:

Incorta URL - the core URL ending in ".com". remove any sub-directories.
Tenant - Can be found as part of your login page for Incorta
Username - Also can be found as part of your login
API Key - see above administrative steps
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
#CREATE FUNCTIONS
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 = requests.post(tokens_url, 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"
    else:
        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
tenant = {INSERT TENANT NAME HERE IN ""}
username = {INSERT USERNAME HERE IN ""}
api_key = {INSERT API KEY HERE IN ""}
incorta_url = {INSERT URL HERE IN ""}
access_token = request_tokens(incorta_url, tenant, username, api_key)
#print(access_token)
headers = {"Authorization": "Bearer "+ access_token, "Accept":"application/json", "Content-type": "application/json"}
#print(headers)
schema_request_payload = {
    "schemaNames":["{SCHEMA NAME}","{SCHEMA NAME}"],
    "loadSource": "FULL",
    "synchronous": True,
    "requestedTimeoutInMilliseconds": 0
}
#print(schema_request_payload)
schema = requests.post(incorta_url+"/incorta/api/v1/schema/load", json=schema_request_payload, headers=headers)
#print(schema.text)