Zuar's ELT+ product Mitto (see why this solution is so epic) has an extensive API, allowing users to remotely interact with jobs, schedules, etc on their Mitto instance. We recently released a Python SDK for interacting with the Mitto API, available on the Python Package Index.

# install with pip
pip install python-mitto-sdk

After logging into your Mitto, you can see a list of API endpoints using the Swagger documentation located at: https://{mitto_url}/docs

At the time of publication, the open source SDK code currently has support for 8 API endpoints. The source code, including examples, is located in the following Git repository: https://github.com/jfreels/python-mitto-sdk

This article will discuss using the SDK to update multiple jobs.

Migrating to a New Input Database

Imagine we're migrating our data from PostgreSQL to MariaDB (hosted on AWS). Currently we have a bunch of SQL jobs that pull data from our PostgreSQL database, and we need to change all those jobs to use the new data in Maria.

Rather than editing each job using the Mitto UI, individually pasting in the new DBO, and saving each one, we can use the Mitto SDK. The following SDK script requires an .env file in the same directory with your Mitto's URL and API Key (available in the settings page of the UI).

# .env
MITTO_BASE_URL=https://{mitto_url}
MITTO_API_KEY=

Setup

At the top of the script we import the necessary packages, including the Mitto SDK, load the .env file and read environment variables, then define the job type, a pattern to match in job DBOs and the new Maria DBO.

import sys
import os
from mitto_sdk import Mitto
from dotenv import load_dotenv

load_dotenv()

BASE_URL = os.getenv("MITTO_BASE_URL")
API_KEY = os.getenv("MITTO_API_KEY")

JOB_TYPE = "io"
INPUT_DBO_LIKE = "postgresql"
NEW_DBO = "mysql+pymysql://<user>:<password>@<hostname>.us-east-1.rds.amazonaws.com:3306/<dbname>"

Get all jobs and loop thru them

Next we instantiate the Mitto class and get all IO jobs, looping thru each job.

def main():
    """change db inputs from postgres to maria"""
    mitto = Mitto(
        base_url=BASE_URL,
        api_key=API_KEY
    )
    jobs = mitto.get_jobs(job_type=JOB_TYPE)
    for job in jobs:

Update jobs

Finally we check each job input DBO looking for the text "postgresql", if a job matches these specifications, we update the job with the new DBO.

        job_id = job["id"]
        job_conf = mitto.get_job(job_id=job_id)
        conf = job_conf["conf"]
        if "input" in conf and "dbo" in conf["input"]:
            if INPUT_DBO_LIKE in conf["input"]["dbo"]:
                conf["input"]["dbo"] = NEW_DBO
                print(f"Updating job conf: {job_id}, {job['title']}")
                mitto.update_job_conf(job_id=job_id, job_conf=conf)

Full script‌‌

"""
This example would update all io jobs with an input
of PostgreSQL to use a new input of MariaDB
"""

import sys
import os
from mitto_sdk import Mitto
from dotenv import load_dotenv

load_dotenv()

BASE_URL = os.getenv("MITTO_BASE_URL")
API_KEY = os.getenv("MITTO_API_KEY")

JOB_TYPE = "io"
INPUT_DBO_LIKE = "postgresql"
NEW_DBO = "mysql+pymysql://<user>:<password>@<hostname>.us-east-1.rds.amazonaws.com:3306/<dbname>"


def main():
    """change db inputs from postgres to maria"""
    mitto = Mitto(
        base_url=BASE_URL,
        api_key=API_KEY
    )
    jobs = mitto.get_jobs(job_type=JOB_TYPE)
    for job in jobs:
        job_id = job["id"]
        job_conf = mitto.get_job(job_id=job_id)
        conf = job_conf["conf"]
        if "input" in conf and "dbo" in conf["input"]:
            if INPUT_DBO_LIKE in conf["input"]["dbo"]:
                conf["input"]["dbo"] = NEW_DBO
                print(f"Updating job conf: {job_id}, {job['title']}")
                mitto.update_job_conf(job_id=job_id, job_conf=conf)
                # break
if __name__ == "__main__":
    sys.exit(main())

NOTE: It is helpful for testing to run the script and only update one job. To do this uncomment # break on the third from the last line.