"""
@author Stefan Schnell <mail@stefan-schnell.de>
@license MIT
@version 1.1
"""
import json
import time
from util.http import Http
class Action:
""" Handles VCF Automation actions
"""
def __init__(self):
self.__http = Http()
def create(
self,
vcoUrl: str,
bearerToken: str,
actionModule: str,
actionName: str,
description: str = "",
version: str = "1.0.0",
outputType: str = "Properties",
inputParameters: list = [],
code: str = "",
runtime: str | None = None,
uniqueName: bool = False
) -> dict:
""" Creates 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 {string} description - Description of the action
@param {string} version - Version of the action
@param {string} outputType - Output type of the action
@param {list} inputParameters - Input parameters of the action
@param {string} code - Script code of the action
@param {string or None} runtime - Runtime environment
@param {bool} uniqueName - Flag to allow multiple actions with
the same display name
@returns {dictionary}
"""
returnValue: dict = {}
try:
url: str = f"{vcoUrl}/api/actions"
if uniqueName:
url += "?uniqueName=true"
else:
url += "?uniqueName=false"
body: dict = {
"module": actionModule,
"name": actionName,
"description": description,
"version": version,
"output-type": outputType,
"input-parameters": inputParameters,
"script": code,
"runtime": runtime
}
returnValue = self.__http.request(
url = url,
bearerToken = bearerToken,
method = "POST",
body = body
)
except Exception as err:
raise ValueError(
f"An error occurred at create action - {err}"
) from err
return returnValue
def read(
self,
vcoUrl: str,
bearerToken: str,
actionModule: str,
actionName: str
) -> dict:
""" Reads 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 {dictionary}
"""
returnValue: dict = {}
try:
actionID: str = self.getActionId(
vcoUrl,
bearerToken,
actionModule,
actionName
)
if actionID:
returnValue = self.__http.request(
url = f"{vcoUrl}/api/actions/{actionID}",
bearerToken = bearerToken
)
except Exception as err:
raise ValueError(
f"An error occurred at read action - {err}"
) from err
return returnValue
def update(
self,
vcoUrl: str,
bearerToken: str,
actionModule: str,
actionName: str,
itemsToChange: list | None = None,
updateReferences: bool = True
) -> None:
""" Upadates 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 {list.<dictionary> or None} itemsToChange - Entries to modify
@param {bool} updateReferences - Flag to update action references
@returns {None}
"""
try:
actionID: str = self.getActionId(
vcoUrl,
bearerToken,
actionModule,
actionName
)
if not actionID:
raise ValueError("Can not find action")
url: str = f"{vcoUrl}/api/actions/{actionID}"
if updateReferences:
url += "?updateReferences=true"
else:
url += "?updateReferences=false"
body: dict = self.read(
vcoUrl = vcoUrl,
bearerToken = bearerToken,
actionModule = actionModule,
actionName = actionName
)
if itemsToChange is None:
itemsToChange = []
for item in itemsToChange:
for key, value in item.items():
body[key] = value
self.__http.request(
url = url,
bearerToken = bearerToken,
method = "PUT",
body = body
)
except Exception as err:
raise ValueError(
f"An error occurred at update action - {err}"
) from err
def delete(
self,
vcoUrl: str,
bearerToken: str,
actionModule: str,
actionName: str,
force: bool = False
) -> None:
""" Deletes 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 {bool} force - Flag to delete action in any case
@returns {None}
"""
try:
actionID: str = self.getActionId(
vcoUrl,
bearerToken,
actionModule,
actionName
)
if not actionID:
raise ValueError("Can not find action")
url: str = f"{vcoUrl}/api/actions/{actionID}"
if force:
url += "?force=true"
else:
url += "?force=false"
self.__http.request(
url = url,
bearerToken = bearerToken,
method = "DELETE"
)
except Exception as err:
raise ValueError(
f"An error occurred at delete action - {err}"
) from err
def getActionId(
self,
vcoUrl: str,
bearerToken: str,
actionModule: str,
actionName: str
) -> str:
""" 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: str = ""
try:
actions = self.__http.request(
url = f"{vcoUrl}/api/actions",
bearerToken = bearerToken
)
if not actions or "link" not in actions:
return returnValue
found: bool = 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(
self,
vcoUrl: str,
bearerToken: str,
actionId: str,
parameters: dict = {}
) -> dict:
""" 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: dict = {}
try:
returnValue = self.__http.request(
url = f"{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(
self,
vcoUrl: str,
bearerToken: str,
executionId: str
) -> dict:
""" 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: dict = {}
try:
returnValue = self.__http.request(
url = (
f"{vcoUrl}/api/actions/{executionId}/"
f"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 invoke(
self,
vcoUrl: str,
bearerToken: str,
actionModule: str,
actionName: str,
parameters: list
) -> dict:
""" 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 {list} parameters - Parameters of the action
@returns {dictionary}
"""
returnValue: dict = {}
try:
actionID: str = self.getActionId(
vcoUrl,
bearerToken,
actionModule,
actionName
)
if not actionID:
raise ValueError("Can not find action")
_parameters: dict = {
"async-execution": False,
"parameters": parameters
}
executionResult: dict = self.__executeAction(
vcoUrl,
bearerToken,
actionID,
_parameters
)
returnValue["executionResult"] = executionResult
if not executionResult or "execution-id" not in executionResult:
raise ValueError(
f"Execution failed or returned invalid response: "
f"{executionResult}"
)
executionId: str = executionResult["execution-id"]
time.sleep(2.5)
actionLog: dict = self.__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 getUsages(
self,
vcoUrl: str,
bearerToken: str,
actionModule: str,
actionName: str
) -> dict:
""" Detects the usages of the 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 {dictionary}
"""
returnValue: dict = {}
try:
actionID = self.getActionId(
vcoUrl,
bearerToken,
actionModule,
actionName
)
if not actionID:
raise ValueError("Can not find action")
returnValue = self.__http.request(
url = f"{vcoUrl}/api/actions/{actionID}/usages",
bearerToken = bearerToken
)
except Exception as err:
raise ValueError(
f"An error occurred at action usages - {err}"
) from err
return returnValue
def getDependencies(
self,
vcoUrl: str,
bearerToken: str,
actionModule: str,
actionName: str
) -> dict:
""" Detects the dependencies of the 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 {dictionary}
"""
returnValue: dict = {}
try:
actionID = self.getActionId(
vcoUrl,
bearerToken,
actionModule,
actionName
)
if not actionID:
raise ValueError("Can not find action")
returnValue = self.__http.request(
url = f"{vcoUrl}/api/actions/{actionID}/dependencies",
bearerToken = bearerToken
)
except Exception as err:
raise ValueError(
f"An error occurred at action dependencies - {err}"
) from err
return returnValue
|
