Ansible Job Monitor

"""
Ansible job monitor

@author Stefan Schnell <mail@stefan-schnell.de>
@version 0.4.0

@param {string} in_userName - Name of the user
@param {SecureString} in_password - Password of the user
@param {string} in_jobTemplateName - Name of the job template
@param {string} in_data - Additional data to send to the server
@param {SecureString} in_sshPassword - Password of the SSH connection
@outputType Properties
"""

import ast
import base64
import json
import ssl
import time
import urllib.request


def ansibleApiCall(
    userName: str,
    userPassword: str,
    apiCall: str,
    method: str = "GET",
    data: str | None = None
) -> dict | bytes | None:

    url: str = "https://yourAnsibleAutomationPlatformURL"
    url += apiCall

    response: http.client.HTTPResponse | None = None
    result: dict | bytes | None = None

    try:

        request = urllib.request.Request(
            url = url,
            method = method,
            data = data
        )

        authorization: str = base64.b64encode(
            bytes(userName + ":" + userPassword, "UTF-8")
        ).decode("UTF-8")
        authorization = "Basic " + authorization

        request.add_header("Authorization", authorization)
        request.add_header("Content-Type", "application/json")

        response = urllib.request.urlopen(
            request,
            context = ssl._create_unverified_context()
        )

        responseCode: int = response.status
        responseRead: bytes = response.read()

        if (
            responseCode == 200 or
            responseCode == 201 or
            responseCode == 202 or
            responseCode == 204
        ):
            if len(responseRead) > 0:
                result = json.loads(responseRead)
            else:
                result = response

    except Exception as err:
        print(err)

    finally:
        if response is not None:
            response.close()

    return result


def jsonStringToDict(
    jsonString: str
) -> dict:

    jsonDict: dict = {}

    try:
        jsonDict = json.loads(jsonString)
    except:
        try:
            jsonDict = ast.literal_eval(jsonString)
        except Exception as err:
            raise Exception("An error occurred at " + \
                "JSON string to dictionary") from err

    return jsonDict


def getIdOfJobTemplate(
    userName: str,
    userPassword: str,
    jobTemplateName: str
) -> int:

    templates: str = str(
        ansibleApiCall(
            userName,
            userPassword,
            "/api/controller/v2/job_templates/?page_size=100"
        )
    )

    templates: dict = jsonStringToDict(templates)

    for template in templates["results"]:
        if (template["name"] == jobTemplateName):
            return template["id"]


def launchJobTemplate(
    userName: str,
    userPassword: str,
    templateId: str,
    data: str | None = None
) -> dict:

    result: str = str(
        ansibleApiCall(
            userName,
            userPassword,
            "/api/controller/v2/job_templates/" + templateId + "/launch/",
            "POST",
            data.encode()
        )
    )

    return jsonStringToDict(result)


def getJobOutput(
    userName: str,
    userPassword: str,
    jobId: str
) -> dict:

    result: str = str(
        ansibleApiCall(
            userName,
            userPassword,
            "/api/controller/v2/jobs/" + jobId + "/stdout/?format=json"
        )
    )

    return jsonStringToDict(result)


def handler(context: dict, inputs: dict) -> dict:

    outputs: dict = {}

    userName: str = inputs["in_userName"]
    userPassword: str = inputs["in_password"]
    nameOfJobTemplate: str = inputs["in_jobTemplateName"]

    jobTemplateId: int = getIdOfJobTemplate(
        userName,
        userPassword,
        nameOfJobTemplate
    )

    # data: str = None
    # if inputs["in_data"]:
    #     data = "{" + inputs["in_data"] + "}"

    data: str = '{"credential_passwords": {"ssh_password": "' + \
        inputs["in_sshPassword"] + '"}}'

    newJob: dict = launchJobTemplate(
        userName,
        userPassword,
        str(jobTemplateId),
        data
    )

    newJobId: int = newJob["id"]

    jobStatus: str | None = None

    while (
        jobStatus == None or
        jobStatus == "running" or
        jobStatus == "pending" or
        jobStatus == "waiting"
    ):

        jobSummary: str = str(ansibleApiCall(
            userName,
            userPassword,
            "/api/controller/v2/jobs/" + str(newJobId)
        ))

        jobSummary: dict = jsonStringToDict(jobSummary)

        jobStatus: str = jobSummary["status"]

        time.sleep(0.125)
        print(".", end = "")

    print("\nJob with Id " + str(newJobId) + " done")

    result: dict = getJobOutput(
        userName,
        userPassword,
        str(newJobId)
    )

    outputs = {
        "status": "done",
        "newJobId": newJobId,
        "result": result["content"]
    }

    return outputs