Example script: Starting a data generation job

Kicking off a data generation using Python. Note the version that this script was tested against. This script is also hosted on Github.

tonic_generation.py
# Note that we do not guarantee backwards compatibility for our API.
# Please look at the Swagger API documentation that comes with your instance
# for your exact version's endpoints and parameters. If you are using hosted
# Tonic, you can find the API documentation here:

# Tested 2021.04.12 with Tonic API v199
import json

# Be sure to install requests via pip, pipenv, or poetry before running.
import requests

# Parameters
TONIC_API_KEY = "<<TONIC API TOKEN>>"
TONIC_URL = "https://<<TONIC HOSTNAME>>"
WORKSPACE_NAME = "<<TONIC WORKSPACE>>"


class TonicSession:
    def __init__(self, base_url, api_key):
        self._base_url = base_url
        self._session = requests.Session()
        self._api_key = api_key
        self._session.headers.update({"Authorization": "Apikey {}".format(api_key)})

    def _get_url(self, api_snippet):
        return "{}{}".format(self._base_url, api_snippet)

    def generate_data(self, workspace_id):
        generate_data_url = self._get_url("/api/generateData/start")
        params = {"workspaceId": workspace_id}

        r = self._session.post(generate_data_url, params=params)

        if r.ok:
            print("Data generation started")
        else:
            r.raise_for_status()

    def generate_data_status(self, workspace_id):
        generate_data_status_url = self._get_url("/api/generateData")
        params = {"workspaceId": workspace_id}

        r = self._session.get(generate_data_status_url, params=params)

        if r.ok:
            print(json.dumps(r.json(), indent=2))
        else:
            r.raise_for_status()

    def get_workspaces(self):
        workspace_list_url = self._get_url("/api/workspace")

        r = self._session.get(workspace_list_url)

        if r.ok:
            return r.json()
        else:
            r.raise_for_status()


def find_workspace_id(workspaces_json, workspace_name):
    for workspace in workspaces_json:
        if workspace["workspaceName"] == workspace_name:
            return workspace["id"]
    raise RuntimeError("No Workspace found with name: {}".format(workspace_name))


def main():
    session = TonicSession(TONIC_URL, TONIC_API_KEY)
    workspace_id = find_workspace_id(session.get_workspaces(), WORKSPACE_NAME)

    # Starts a new data generation
    session.generate_data(workspace_id)

    # Prints out the current status of the workspace
    session.generate_data_status(workspace_id)


if __name__ == "__main__":
    main()

Last updated