restcurluac
List all dependencies of a workflow in UAC / StoneBranch
I have a workflow in UAC that contains inside it another list of workflows and each one of it a series of tasks.
I want to make an API Call to list all the WFs and tasks and their dependencies. Is that possible?
I only managed to extract the first level of information.
Structure of workflow:
INITIAL_WF:
|_Inside_WF_1
| |
| |_ task1
| |
| |_task2
|
|_Inside_WF_2
|
|_task1
|
|_task2
|
|_Inside_WF_2_1
|
|_ task1
API CALLS:
curl --request GET http://UAC_HOST/uc/resources/workflow/vertices?workflowname=INITIAL_WF --header "Content-Type: application/json" --header "Authorization: {TOKEN}"
This command will only give me information of Inside_WF_1 and Inside_WF_2.
Solution
In the UAC UI under the Right Click "Workflow Task Commands" you can select View Tree to display the following:
View Tree Report:
However this cannot currently be printed or accessed with a single API call. This is under consideration for a future release.
In the mean time I have a Python script that does the equivalent:
python .\Controller_WorkflowTree.py --controllerurl https://dev-uac:8499/uc --username ccocksedge --password sbus77 --workflow '#app_collections ${ao_schedule_date}'
WORKFLOW = #app_collections ${ao_schedule_date}
Task Name Task Alias Task Type
Pull_Todays_Inputs None Stored Procedure
Check_Space_B4_Load None System Monitor
Daily_Returns None Workflow
Task Name Task Alias Task Type
Daily_Calc_Returns None Windows
Report_Inputs None Windows
Report_Vendors None Windows
Report_Returns None Windows
Weekly_Returns None Workflow
Task Name Task Alias Task Type
Weekly_Calc_Returns None Windows
Report_Inputs None Windows
Report_Vendors None Windows
Report_Returns None Windows
Monthly_Returns None Workflow
Task Name Task Alias Task Type
Monthly_Calc_Returns None Windows
Report_Inputs None Windows
Report_Returns None Windows
Report_Vendors None Windows
Disk_Clean_Up None Windows
List_Receipts None Windows
Load_Results None Stored Procedure
Detect_Vendor_Feed None Remote File Monitor
Download_Vendor_Feeds None File Transfer
Here is the code:
#!/opt/universal/python/bin/python3
# --
# Origins: Stonebranch
# Author: Colin Cocksedge colin.cocksedge@stonebranch.com
# Date: 22-APR-2020
#
# Copyright (c) Stonebranch, 2020. All rights reserved.
#
# Purpose: Workflow Tree Report
#
version = "1.0"
# Version History: 1.0 CCO 22-APR-2020 Initial Version
# --
# -- Main Logic Function
def main():
ScriptSetup() # -- Import Required Modules, Setup Logging Format, Set Variables
indent = 0 # -- Set base indent for formatted tree report
print("WORKFLOW = {0}".format(args.workflow))
WorkflowReport(args.workflow,indent) # -- Workflow Tree Report
# --
# -- Import Required Modules, Setup Logging Format, Set Variables
def ScriptSetup():
# -- Import required python modules
import argparse, logging, sys, urllib
# -- Import requests module, if error then install requests
try:
import requests
except:
try:
from pip import main as pipmain
except:
from pip._internal import main as pipmain
pipmain(
[
"install",
"--trusted-host=pypi.python.org",
"--trusted-host=pypi.org",
"--trusted-host=files.pythonhosted.org",
"requests",
]
)
global argparse, logging, sys, requests, urllib
# -- Set Variables
parser=argparse.ArgumentParser(description='Purpose : Workflow Tree Report')
# ## -->
parser.add_argument("--controllerurl", default="http://localhost:8080/opswise")
parser.add_argument("--username", default="ops.admin")
parser.add_argument("--password", default="secret")
parser.add_argument("--workflow", default="My_Workflow")
# ## -->
parser.add_argument("--loglevel", default="None")
global args
args=parser.parse_args()
# -- Setup Logging
numeric_level = getattr(logging, args.loglevel.upper(), None)
logging.basicConfig(format='%(asctime)-15s - %(levelname)-8s - %(message)s', level=numeric_level)
# -- Print Paramater Values
logging.debug("Executing version {0} with the following parameters : {1}".format(version, args))
# -- Strip Trailing / from Remote Controller URL to Avoid 404 Resource Not Found
args.controllerurl = args.controllerurl.rstrip("/")
# -- Ignore Https Warnings
import urllib3
urllib3.disable_warnings()
# --
# ## --> Functions Go Here
# -- Workflow Tasks
def WorkflowReport(workflow,indent):
WorkflowTaskList = GetWorkflowTasks(workflow) # Find All Tasks in Workflow
print("{0} {1:50} {2:50} {3:30}".format(" "*indent,"Task Name","Task Alias","Task Type"))
for task in WorkflowTaskList:
# For Each Workflow Task
taskalias = task['alias']
taskname = task['task']['value']
# Get the Task Type
tasktype = GetTaskType(taskname)
print("{0} {1:50} {2:50} {3}".format(" "*indent,taskname,str(taskalias),tasktype))
# If Task Type = Workflow Then process the subworkflow's tasks
if tasktype == "Workflow":
indent = indent + 1 # increment subworkflows tree report level
WorkflowReport(taskname,indent)
indent = indent - 1 # decrement tree report level
# --
# -- Get All Tasks in Workflow
def GetWorkflowTasks(workflowname):
logging.debug("Call resources/workflow/vertices API")
payload = {'workflowname':workflowname}
query = urllib.parse.urlencode(payload)
ListWorkflowTasks = apicall(args.username, args.password, "get", "/workflow/vertices?", query, "")
WorkflowTasks = ListWorkflowTasks.json()
return(WorkflowTasks)
# --
# -- Get Task Type
def GetTaskType(taskname):
logging.debug("Call resources/task/list API")
data = {
"name": taskname
}
ListTask = apicall(args.username, args.password, "post", "/task/list", "None", data)
tasklist = ListTask.json()
tasktype = tasklist[0]['type']
return(tasktype)
# --
# -- Call UAC API
def apicall(username, password, method, resource, query, jsondata):
if str(query) == "None":
query=""
headers = {
'content-type': 'application/json',
'Accept': 'application/json'
}
uri="{0}/resources{1}{2}".format(args.controllerurl, resource, query)
try:
if str(method) == "get":
response = requests.get(uri, headers=headers, auth=(username, password), verify=False)
if str(method) == "post":
response = requests.post(uri, headers=headers, json=jsondata, auth=(username, password), verify=False)
except:
logging.error("Error Calling {0} API {1}".format(args.controllerurl, sys.exc_info()))
sys.exit(1)
if(response.ok):
pass
else:
logging.critical("{0} Error Code : {1}".format(uri, response.status_code))
logging.critical("Failed with reason : {}".format(response.text))
sys.exit(1)
return(response)
# --
# -->
# -- Execute
main()
- How to hide NAs when using xlsx::saveWorkbook?
- How do I retrieve a simple numeric value from a named numeric vector in R?
- Matching pair-wise columns from left to right across rows in one dataframe to another dataframe and adding new columns with matching values
- Income to outcome flow chart in Sankey plotly R
- color mapping in geom_conn_bundle not showing correctly
- Print R package startup message AFTER automatic package conflict messages instead of before
- Summing a set of R dataframe rows (column-wise), while retaining the first n columns
- Added variable / partial regression plots for groups in an interaction?
- how to make a topoplot in R with coordinates variable distribution
- List of all functions in base R?
- Plotting multiple plots for different initial conditions in one graph
- Printing repetitively on the same line in R
- Generating UI/Server based on initial selection
- Subset dataframe based on pickerInput
- How to let user pick the data in R-shiny?
- Couldn't show my simple bar charts separately on Shiny R dashboardBody
- How to programmatically filter contents of a second shiny app displayed via iframe
- How to select specific interesting groups for the boxplot in R Shiny app?
- Crosstable and Plot grouping with reactive values
- Is there a way to make multiple Shiny picker inputs where the selections must be disjoint?
- Delay/avoid duplication of shiny server side functions until after credentials
- Predictions only returns value "1"
- How to display a busy indicator in a shiny app?
- Append doesn't work when writing to CSV in R
- Changing the start date of a gantt chart in DiagrammeR
- Check for installed packages before running install.packages()
- Compare two columns element-wise
- For loop is several times faster in R than in Python using the rpy2 library
- Setting a flag based on two samples' dates
- How to bring pop-ups for long text string selection items forward in front of modal they are rendered from when using virtualSelectInput()?