WALKOFF
WALKOFF copied to clipboard
api tutorial isn't working
I copy and pasted the api tutorial stuff in to a python file and get this output when run
PS C:\Users\aoighost\Desktop\motokodev> python .\walkoffapitest.py
Authenticating...
Enter password for WALKOFF user admin@http://127.0.0.1:5000:
Traceback (most recent call last):
File ".\walkoffapitest.py", line 200, in
OK, I fixed it under linux. Not sure if this will work in windows as windows is broken. The tutorial needs to be updated. Also should note this uses unencrypted. Had issues with the self signed cert.
`import argparse import requests import getpass import time
class APITutorial: def init(self, address, username): self.address = address self.username = username self.refresh_token = None self.auth_header = None self.is_connected = False self.url = "http://localhost:8081"
def prompt_pass(self):
return getpass.getpass(
prompt="Enter password for WALKOFF user {}@{}: ".format(
self.username, self.address))
def authenticate(self):
print("Authenticating...")
r = requests.post(self.address + "/api/auth",
json=dict(username=self.username,
password=self.prompt_pass()),
timeout=2)
if r.status_code == 201:
r = r.json()
self.refresh_token = r['refresh_token']
self.auth_header = {'Authorization': 'Bearer {}'.format(
r['access_token'])}
self.is_connected = True
print("Authenticated.")
elif r.status_code == 401:
print("Authentication Error.")
elif r.status_code == 404:
print("WALKOFF instance not found.")
else:
print("Unknown response: {}: {}".format(
r.status_code, r.json()))
return self.is_connected
def refresh(self):
refreshed = False
if self.is_connected:
print("Refreshing access token..."),
refresh_header = {'Authorization': 'Bearer {}'.format(
self.refresh_token)}
r = requests.post(self.address + "/api/auth/refresh",
headers=refresh_header,
timeout=2)
if r.status_code == 201:
r = r.json()
self.auth_header = {'Authorization': 'Bearer {}'.format(
r['access_token'])}
print("Access token refreshed.")
refreshed = True
elif r.status_code == 401:
print("Authentication Error.")
elif r.status_code == 404:
print("WALKOFF instance not found.")
else:
print("Unknown response: {}: {}".format(
r.status_code, r.json()))
return refreshed
def logout(self):
logged_out = False
if self.is_connected:
print("Logging out..."),
r = requests.post(self.address + "/api/auth/logout",
headers=self.auth_header,
json=dict(refresh_token=self.refresh_token),
timeout=2)
if r.status_code == 204:
self.is_connected = False
print("Logged out.")
logged_out = True
elif r.status_code == 400:
print("Bad or missing refresh token.")
else:
print("Unknown response: {}: {}".format(
r.status_code, r.json()))
return logged_out
def get_workflow_by_name(self, workflow_name):
"""" Attempts to find a workflow by name and returns its id """
if self.is_connected:
r = requests.get(self.url + "/api/workflows",
headers=self.auth_header)
workflows = r.json()
for workflow in workflows:
if workflow["name"].lower() == workflow_name.lower():
return workflow["id_"]
print("Workflow: {} does not exist.".format(workflow_name))
def check_workflow_exists(self, workflow_name):
workflow_id = self.get_workflow_by_name(workflow_name)
if self.is_connected and workflow_id is not None:
print("Querying workflow: {}\t with ID:{}...".format(
workflow_name, workflow_id)),
ep = "/api/workflows/{}".format(workflow_id)
r = requests.get(self.url + ep,
headers=self.auth_header,
timeout=2)
if r.status_code == 200:
print("Workflow: {}\t{} exists".format(workflow_name,
workflow_id))
return workflow_id
elif r.status_code == 404:
print("Workflow does not exist.")
else:
print("Unknown response: {}: {}".format(r.status_code,
r.json()))
def execute_workflow(self, workflow_id):
execution_uid = None
if self.is_connected:
print("Executing workflow..."),
ep = "/api/workflowqueue"
r = requests.post(self.url + ep, headers=self.auth_header,
json={"workflow_id": workflow_id},
timeout=2)
if r.status_code == 202:
print(r.json())
execution_uid = r.json()["execution_id"]
print("Workflow: {} was executed with ID: {}".format(
workflow_id, execution_uid))
elif r.status_code == 404:
print("Workflow does not exist.")
else:
print("Unknown response: {}: {}".format(r.status_code,
r.json()))
return execution_uid
def check_result(self, execution_uid):
success = False
if self.is_connected:
print("Checking results..."),
ep = "/api/workflowqueue/{}".format(execution_uid)
r = requests.get(self.url + ep, headers=self.auth_header,
timeout=2)
if r.status_code == 200:
success = True
print("Results of {}: {}".format(
execution_uid, r.json()))
else:
print("Unknown response: {}: {}".format(
r.status_code, r.json()))
return success
def main(): parser = argparse.ArgumentParser() parser.add_argument("-a", "--address", type=str, default="http://127.0.0.1:8081") parser.add_argument("-u", "--username", type=str, default="admin") args = parser.parse_args()
i = APITutorial(args.address, args.username)
i.authenticate()
i.refresh()
workflow_id = i.check_workflow_exists("Tutorial")
euid = i.execute_workflow(workflow_id)
time.sleep(3)
i.check_result(euid)
i.logout()
if name == "main": main()`
Thanks for taking a look at that.
That tutorial will eventually be replaced with one that will make use of the client libraries we generate with OpenAPI Generator. An initial version of the Python client package is in the common directory with a brief example. Hopefully these will make it easier to use Walkoff's REST API, but there are customizations we need to iron out and OpenAPI spec discrepancies to fix before I'm confident in recommending its usage.