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 [email protected]
# 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()
- Meteor server api using iron router returns HTML not the response
- How to return global object from a Rocket REST API?
- How to map this Json with rest template spring boot
- Office 365 REST API - Creating a Contact gives me HTTPCode 400
- Guidance for writing a wrapper for a REST API
- Django rest framework where to write complex logic in serializer.py or views.py?
- REST and spring-mvc
- Cookie in Thunder Client VS Code extension
- How to write a generic handler in Go?
- How to make a script to automate taking the GET data from my wordpress site and POST to another server
- How do you set a scenario when doing a restful call in Yii2 to return certain fields
- Spring - checking path variable for null at controller
- How to transform Model-View-Presenter architecture to WebAPI?
- Go file server doesn't serve folder
- How to use jti claim in a JWT
- Rewrite POST method on rest api yii2
- JFrog REST API to Export the X-ray SBOM report
- How can I get BizTalk to allow a large array in querystring for consuming a REST service?
- Why am i getting 405 Method Not Allowed while doing a POST Request
- Use CakePHP Http Client with Magento2 rest API search criteria
- API Key in path
- Trying to call rest APIs, but get http request exception
- Invoke-WebRequest, POST with parameters
- An established connection was aborted by the software in your host machine tomcat jackson
- Empty Form When Initializing from Rest API
- 404-not-found-while-running-spring-boot-rest-api
- Pass array as query parameter in Postman
- Where should the filtering logic be: in the frontend or in the backend?
- Can I get the company name/logo with the LinkedIn API?
- Nuxt 3 folder structure for API call functions