"""
@author Stefan Schnell <mail@stefan-schnell.de>
@license MIT
@version 1.1
"""
import json
import time
from .http import Http
from .category import Category
class Workflow:
""" Handles VCF Automation workflows
"""
def __init__(
self
) -> None:
self.__http = Http()
self.__category = Category()
def create(
self,
vcoUrl: str,
bearerToken: str,
workflowFolder: str,
workflowName: str,
description: str = "",
version: str = "1.0.0",
variables: list[dict] | None = None,
inputs: dict | None = None,
outputs: dict | None = None,
schema: list[dict] | None = None,
uniqueName: bool = False
) -> None:
""" Creates a workflow
@param {string} vcoUrl - URL of Aria orchestrator
@param {string} bearerToken
@param {string} workflowFolder - Folder of the workflow
@param {string} workflowName - Name of the workflow
@param {string} description - Description of the workflow
@param {string} version - Version of the workflow
@param {list.<dictionary> or None} variables
@param {dictionary or None} inputs
@param {dictionary or None} outputs
@param {list.<dictionary> or None} schema
@param {bool} uniqueName - Flag to allow multiple workflows with
the same display name
"""
_variables: list[dict] = []
if variables is not None and isinstance(variables, list):
_variables = variables
_inputs: dict = { "param": [] }
if inputs is not None and isinstance(inputs, dict):
_inputs = inputs
_outputs: dict = { "param": [] }
if outputs is not None and isinstance(outputs, dict):
_outputs = outputs
_schema: list[dict] = [
{
"position": { "x": 100, "y": 120 },
"name": "item0",
"type": "end",
"end-mode": "0",
"comparator": 0
}
]
if schema is not None and isinstance(schema, list):
_schema = schema
try:
categoryId: str = self.__category.getId(
vcoUrl = vcoUrl,
bearerToken = bearerToken,
categoryType = "WorkflowCategory",
categoryPath = workflowFolder
)
if categoryId:
url: str = f"{vcoUrl}/api/workflows"
if uniqueName:
url += "?uniqueName=true"
else:
url += "?uniqueName=false"
# The creation is an instantiation and only the name and
# category-id are processed, all other keys are ignored.
# That is why it is necessary to proceed in two steps.
# First the creation, then an update.
body: dict = {
"category-id": categoryId,
"name": workflowName,
}
returnValue = self.__http.request(
url = url,
bearerToken = bearerToken,
method = "POST",
body = body
)
time.sleep(2.5)
self.update(
vcoUrl = vcoUrl,
bearerToken = bearerToken,
workflowFolder = workflowFolder,
workflowName = workflowName,
itemsToChange = [
{ "description": description },
{ "version": version },
{ "attrib": _variables },
{ "input": _inputs },
{ "output": _outputs },
{ "workflow-item": _schema }
]
)
except Exception as err:
raise ValueError(
f"An error occurred at create workflow - {err}"
) from err
def read(
self,
vcoUrl: str,
bearerToken: str,
workflowFolder: str,
workflowName: str
) -> dict:
""" Reads a workflow
@param {string} vcoUrl - URL of Aria orchestrator
@param {string} bearerToken
@param {string} workflowFolder - Folder of the workflow
@param {string} workflowName - Name of the workflow
@returns {dictionary}
"""
returnValue: dict = {}
try:
workflowId = self.getWorkflowId(
vcoUrl,
bearerToken,
workflowFolder,
workflowName
)
if workflowId:
returnValue = self.__http.request(
url = f"{vcoUrl}/api/workflows/{workflowId}",
bearerToken = bearerToken
)
except Exception as err:
raise ValueError(
f"An error occurred at read workflow - {err}"
) from err
return returnValue
def update(
self,
vcoUrl: str,
bearerToken: str,
workflowFolder: str,
workflowName: str,
itemsToChange: list | None = None
) -> None:
""" Updates a workflow
@param {string} vcoUrl - URL of Aria orchestrator
@param {string} bearerToken
@param {string} workflowFolder - Folder of the workflow
@param {string} workflowName - Name of the workflow
@param {list.<dictionary> or None} itemsToChange - Entries to modify
"""
try:
workflowId = self.getWorkflowId(
vcoUrl,
bearerToken,
workflowFolder,
workflowName
)
if not workflowId:
raise ValueError("Can not find workflow")
body: dict = self.getSchema(
vcoUrl,
bearerToken,
workflowFolder,
workflowName
)
if itemsToChange is None:
itemsToChange = []
for item in itemsToChange:
for key, value in item.items():
body[key] = value
self.__http.request(
url = f"{vcoUrl}/api/workflows/{workflowId}/content",
bearerToken = bearerToken,
method = "PUT",
body = body
)
except Exception as err:
raise ValueError(
f"An error occurred at update workflow - {err}"
) from err
def delete(
self,
vcoUrl: str,
bearerToken: str,
workflowFolder: str,
workflowName: str,
force: bool = False,
forceDeleteLocked = False
) -> None:
""" Deletes a workflow
@param {string} vcoUrl - URL of Aria orchestrator
@param {string} bearerToken
@param {string} workflowFolder - Folder of the workflow
@param {string} workflowName - Name of the workflow
@param {bool} force - Flag to delete workflow in any case
@param {bool} forceDeleteLocked - Flag to delete workflow even
locks exists in database
@returns {None}
"""
try:
workflowId = self.getWorkflowId(
vcoUrl,
bearerToken,
workflowFolder,
workflowName
)
if not workflowId:
raise ValueError("Can not find workflow")
url: str = f"{vcoUrl}/api/workflows/{workflowId}"
if force:
url += "?force=true"
else:
url += "?force=false"
if forceDeleteLocked:
url += "&forceDeleteLocked=true"
else:
url += "&forceDeleteLocked=false"
self.__http.request(
url = url,
bearerToken = bearerToken,
method = "DELETE"
)
except Exception as err:
raise ValueError(
f"An error occurred at delete workflow - {err}"
) from err
def getWorkflowId(
self,
vcoUrl: str,
bearerToken: str,
workflowFolder: str,
workflowName: str
) -> str:
""" Gets the ID of a workflow.
@param {string} vcoUrl - URL of Aria orchestrator
@param {string} bearerToken
@param {string} workflowFolder - Folder of the workflow
@param {string} workflowName - Name of the workflow
@returns {string}
"""
returnValue: str = ""
try:
workflows = self.__http.request(
url = vcoUrl + "/api/workflows",
bearerToken = bearerToken
)
found: bool = False
for workflow in workflows["link"]:
for attribute in workflow["attributes"]:
if attribute["name"] == "name" and \
attribute["value"] == workflowName:
for attribute in workflow["attributes"]:
if attribute["name"] == "categoryId":
categoryId = attribute["value"]
categoryPath = self.__category.getPath(
vcoUrl,
bearerToken,
categoryId
)
if categoryPath == workflowFolder:
for attribute in workflow["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 workflow ID - {err}"
) from err
return returnValue
def __executeWorkflow(
self,
vcoUrl: str,
bearerToken: str,
workflowId: str,
parameters: dict = {}
) -> dict:
""" Executes a workflow.
@param {string} vcoUrl - URL of Aria orchestrator
@param {string} bearerToken
@param {string} workflowId - ID of the workflow
@param {dictionary} parameters - Parameters of the workflow
@returns {dictionary}
"""
returnValue: dict = {}
try:
returnValue = self.__http.request(
url = (
f"{vcoUrl}/api/workflows/{workflowId}/executions"
),
bearerToken = bearerToken,
method = "POST",
body = parameters
)
except Exception as err:
raise ValueError(
f"An error occurred at workflow executing - {err}"
) from err
return returnValue
def __getWorkflowLog(
self,
vcoUrl: str,
bearerToken: str,
workflowId: str,
executionId: str
) -> dict:
""" Delivers the workflow run log.
@param {string} vcoUrl - URL of Aria orchestrator
@param {string} bearerToken
@param {string} workflowId - ID of the workflow
@param {string} executionId - ID of the execution, from
executeWorkflow
@returns {dictionary}
"""
returnValue: dict = {}
try:
returnValue = self.__http.request(
url = (
f"{vcoUrl}/api/workflows/{workflowId}/executions/"
f"{executionId}/logs?maxResult=2147483647"
),
bearerToken = bearerToken
)
except Exception as err:
raise ValueError(
f"An error occurred at get workflow log - {err}"
) from err
return returnValue
def __getWorkflowSyslog(
self,
vcoUrl: str,
bearerToken: str,
workflowId: str,
executionId: str
) -> dict:
""" Delivers the workflow scripting and system log.
@param {string} vcoUrl - URL of Aria orchestrator
@param {string} bearerToken
@param {string} workflowId - ID of the workflow
@param {string} executionId - ID of the execution, from
executeWorkflow
@returns {dictionary}
"""
returnValue: dict = {}
try:
returnValue = self.__http.request(
url = (
f"{vcoUrl}/api/workflows/{workflowId}/executions/"
f"{executionId}/syslogs?maxResult=2147483647"
),
bearerToken = bearerToken
)
except Exception as err:
raise ValueError(
f"An error occurred at get workflow syslog - {err}"
) from err
return returnValue
def __getWorkflowState(
self,
vcoUrl: str,
bearerToken: str,
workflowId: str,
executionId: str
) -> str:
""" Delivers the running state of a workflow.
@param {string} vcoUrl - URL of Aria orchestrator
@param {string} bearerToken
@param {string} workflowId - ID of the workflow
@param {string} executionId - ID of the execution, from
executeWorkflow
@returns {string}
"""
returnValue: str = ""
try:
returnValue = self.__http.request(
url = (
f"{vcoUrl}/api/workflows/{workflowId}/executions/"
f"{executionId}/state"
),
bearerToken = bearerToken
)["value"]
except Exception as err:
raise ValueError(
f"An error occurred at get workflow state - {err}"
) from err
return returnValue
def __getOutputParameters(
self,
vcoUrl: str,
bearerToken: str,
workflowId: str,
executionId: str
) -> dict:
""" Delivers the output parameters of the workflow run.
@param {string} vcoUrl - URL of Aria orchestrator
@param {string} bearerToken
@param {string} workflowId - ID of the workflow
@param {string} executionId - ID of the execution, from
executeWorkflow
@returns {dictionary}
"""
returnValue: dict = {}
try:
returnValue = self.__http.request(
url = (
f"{vcoUrl}/api/workflows/{workflowId}/executions/"
f"{executionId}"
),
bearerToken = bearerToken
)["output-parameters"]
except Exception as err:
raise ValueError(
f"An error occurred at get workflow syslog - {err}"
) from err
return returnValue
def invoke(
self,
vcoUrl: str,
bearerToken: str,
workflowFolder: str,
workflowName: str,
parameters: dict
) -> dict:
""" Calls a workflow
@param {string} vcoUrl - URL of Aria orchestrator
@param {string} bearerToken
@param {string} workflowFolder - Folder of the workflow
@param {string} workflowName - Name of the workflow
@param {dictionary} parameters - Parameters of the workflow
@returns {dictionary}
"""
returnValue: dict = {}
try:
workflowId = self.getWorkflowId(
vcoUrl,
bearerToken,
workflowFolder,
workflowName
)
if not workflowId:
raise ValueError("Can not find workflow")
executionResult = self.__executeWorkflow(
vcoUrl,
bearerToken,
workflowId,
parameters
)
executionId = executionResult["id"]
while True:
workflowState = self.__getWorkflowState(
vcoUrl,
bearerToken,
workflowId,
executionId
)
if workflowState == "completed":
break
else:
time.sleep(2)
outputParameters = self.__getOutputParameters(
vcoUrl,
bearerToken,
workflowId,
executionId
)
workflowLog = self.__getWorkflowLog(
vcoUrl,
bearerToken,
workflowId,
executionId
)["logs"]
workflowSyslog = self.__getWorkflowSyslog(
vcoUrl,
bearerToken,
workflowId,
executionId
)["logs"]
returnValue["outputParameters"] = outputParameters
returnValue["runLogs"] = workflowLog
returnValue["sysLogs"] = workflowSyslog
except Exception as err:
raise ValueError(
f"An error occurred at call workflow - {err}"
) from err
return returnValue
def getUsages(
self,
vcoUrl: str,
bearerToken: str,
workflowFolder: str,
workflowName: str
) -> dict:
""" Detects the usages of the workflow
@param {string} vcoUrl - URL of Aria orchestrator
@param {string} bearerToken
@param {string} workflowFolder - Folder of the workflow
@param {string} workflowName - Name of the workflow
@returns {dictionary}
"""
returnValue: dict = {}
try:
workflowId = self.getWorkflowId(
vcoUrl,
bearerToken,
workflowFolder,
workflowName
)
if not workflowId:
raise ValueError("Can not find workflow")
returnValue = self.__http.request(
url = f"{vcoUrl}/api/workflows/{workflowId}/usages",
bearerToken = bearerToken
)
except Exception as err:
raise ValueError(
f"An error occurred at workflow usages - {err}"
) from err
return returnValue
def getDependencies(
self,
vcoUrl: str,
bearerToken: str,
workflowFolder: str,
workflowName: str
) -> dict:
""" Detects the dependencies of the workflow
@param {string} vcoUrl - URL of Aria orchestrator
@param {string} bearerToken
@param {string} workflowFolder - Folder of the workflow
@param {string} workflowName - Name of the workflow
@returns {dictionary}
"""
returnValue: dict = {}
try:
workflowId = self.getWorkflowId(
vcoUrl,
bearerToken,
workflowFolder,
workflowName
)
if not workflowId:
raise ValueError("Can not find workflow")
returnValue = self.__http.request(
url = (
f"{vcoUrl}/api/workflows/{workflowId}/dependencies"
),
bearerToken = bearerToken
)
except Exception as err:
raise ValueError(
f"An error occurred at workflow dependencies - {err}"
) from err
return returnValue
def getSchema(
self,
vcoUrl: str,
bearerToken: str,
workflowFolder: str,
workflowName: str
) -> dict:
""" Gets the schema of the workflow
@param {string} vcoUrl - URL of Aria orchestrator
@param {string} bearerToken
@param {string} workflowFolder - Folder of the workflow
@param {string} workflowName - Name of the workflow
@returns {dictionary}
"""
returnValue: dict = {}
try:
workflowId = self.getWorkflowId(
vcoUrl,
bearerToken,
workflowFolder,
workflowName
)
if not workflowId:
raise ValueError("Can not find workflow")
returnValue = self.__http.request(
url = f"{vcoUrl}/api/workflows/{workflowId}/content",
bearerToken = bearerToken
)
except Exception as err:
raise ValueError(
f"An error occurred at workflow schema - {err}"
) from err
return returnValue
|