"""
@module de.stschnell
@version 0.2.0
@runtime python:3.10
@memoryLimit 128000000
@inputType in_userName {string}
@inputType in_password {string}
@outputType Properties
"""
import base64
import json
import ssl
import time
import urllib.request
def request(
url,
user = None,
password = None,
bearerToken = None,
method = "GET",
body = {},
contentType = "application/json;charset=utf-8",
accept = "application/json"
):
""" Executes a REST request
@param {string} url - URL to execute the request
@param {string} user
@param {string} password
@param {string} bearerToken
@param {string} method - Method of request, e.g. GET, POST, etc
@param {dictionary} body - Body of request
@param {string} contentType - MIME type of the body for the request
@param {string} accept - MIME type of content expect/prefer as response
@returns {dictionary or bytes}
"""
returnValue = {}
try:
request = urllib.request.Request(
url = url,
method = method,
data = bytes(json.dumps(body).encode("utf-8"))
)
if user and password:
authorization = base64.b64encode(
bytes(user + ":" + password, "UTF-8")
).decode("UTF-8")
request.add_header(
"Authorization", "Basic " + authorization
)
if bearerToken:
request.add_header(
"Authorization", "Bearer " + bearerToken
)
request.add_header(
"Content-Type", contentType
)
request.add_header(
"Accept", accept
)
response = urllib.request.urlopen(
request,
context = ssl._create_unverified_context()
)
if response.getcode() == 200 or response.getcode() == 202:
if "json" in accept:
returnValue = json.loads(response.read())
else:
returnValue = response.read()
except Exception as err:
raise Exception(f"An error occurred at request - {err}") \
from err
return returnValue
def getActionId(vcoUrl, bearerToken, actionModule, actionName):
""" Gets the ID of an action.
@param {string} vcoUrl - URL of Aria orchestrator
@param {string} bearerToken
@param {string} actionModule - Module of the action
@param {string} actionName - Name of the action
@returns {string}
"""
returnValue = ""
try:
actions = request(
url = vcoUrl + "/api/actions",
bearerToken = bearerToken
)
found = False
for action in actions["link"]:
for attribute in action["attributes"]:
if attribute["name"] == "fqn" and \
attribute["value"] == actionModule + "/" + actionName:
for attribute in action["attributes"]:
if attribute["name"] == "id":
returnValue = attribute["value"]
found = True
if found:
break
except Exception as err:
raise ValueError(f"An error occurred at get Action ID - {err}") \
from err
return returnValue
def executeAction(vcoUrl, bearerToken, actionId, parameters = {}):
""" Executes an action.
@param {string} vcoUrl - URL of Aria orchestrator
@param {string} bearerToken
@param {string} actionId - ID of the action
@param {dictionary} parameters - Parameters of the action
@returns {dictionary}
"""
returnValue = {}
try:
returnValue = request(
url = vcoUrl + "/api/actions/" + actionId + "/executions",
bearerToken = bearerToken,
method = "POST",
body = parameters
)
except Exception as err:
raise ValueError(f"An error occurred at action executing - {err}") \
from err
return returnValue
def getActionLog(vcoUrl, bearerToken, executionId):
""" Delivers the action log.
@param {string} vcoUrl - URL of Aria orchestrator
@param {string} bearerToken
@param {string} executionId - ID of the execution, from executeAction
@returns {dictionary}
"""
returnValue = {}
try:
returnValue = request(
url = vcoUrl + "/api/actions/" + executionId + \
"/logs?maxResult=2147483647",
bearerToken = bearerToken
)
except Exception as err:
raise ValueError(f"An error occurred at get action log - {err}") \
from err
return returnValue
def callAction(vcoUrl, bearerToken, actionModule, actionName, parameters):
""" Calls an action
@param {string} vcoUrl - URL of Aria orchestrator
@param {string} bearerToken
@param {string} actionModule - Module of the action
@param {string} actionName - Name of the action
@param {dictionary} parameters - Parameters of the action
@returns {dictionary}
"""
returnValue = {}
try:
actionID = getActionId(
vcoUrl,
bearerToken,
actionModule,
actionName
)
_parameters = {
"async-execution": False,
"parameters": parameters
}
executionResult = executeAction(
vcoUrl,
bearerToken,
actionID,
_parameters
)
returnValue["executionResult"] = executionResult
executionId = executionResult["execution-id"]
time.sleep(2.5)
actionLog = getActionLog(vcoUrl, bearerToken, executionId)
returnValue["actionLog"] = actionLog
except Exception as err:
raise ValueError(f"An error occurred at call action - {err}") \
from err
return returnValue
def handler(context, inputs):
""" Aria Automation standard handler, the main function.
"""
vcoUrl = context["vcoUrl"]
bearerToken = context["getToken"]()
output = {}
try:
# Begin of action call -----------------------------------------
# The setting of the parameters is described in the
# REST API Calling Conventions.
# @example
# actionName = "getDefaultCompanyName"
# actionModule = "com.vmware.constants"
# parameters = []
# @example
actionName = "getFileName"
actionModule = "com.vmware.basic"
parameters = [
{
"name": "fileName",
"type": "Path",
"value": {
"string": {
"value": "/stefan/test/bambi"
}
}
}
]
output = callAction(
vcoUrl, bearerToken, actionModule, actionName, parameters
)
# print(output["executionResult"]["value"]["string"]["value"])
# delivers VMware Inc. with the first example and
# bambi with the second example
# End of action call -------------------------------------------
outputs = {
"status": "done",
"error": None,
"result": output
}
except Exception as err:
outputs = {
"status": "incomplete",
"error": repr(err),
"result": output
}
return outputs
|
