api tutorial isn't working
peterclemenko opened this issue · 2 comments
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
main()
File ".\walkoffapitest.py", line 190, in main
i.authenticate()
File ".\walkoffapitest.py", line 25, in authenticate
timeout=2)
File "C:\Python27\lib\site-packages\requests\api.py", line 116, in post
return request('post', url, data=data, json=json, **kwargs)
File "C:\Python27\lib\site-packages\requests\api.py", line 60, in request
return session.request(method=method, url=url, **kwargs)
File "C:\Python27\lib\site-packages\requests\sessions.py", line 533, in request
resp = self.send(prep, **send_kwargs)
File "C:\Python27\lib\site-packages\requests\sessions.py", line 646, in send
r = adapter.send(request, **kwargs)
File "C:\Python27\lib\site-packages\requests\adapters.py", line 516, in send
raise ConnectionError(e, request=request)
requests.exceptions.ConnectionError: HTTPConnectionPool(host='127.0.0.1', port=5000): Max retries exceeded with url: /api/auth (Caused by NewConnectionError('<urllib3.connection.HTTPConnection object at 0x0000000003A502E8>: Failed to establish a new connection: [Errno 10061] No connection could be made because the target machine actively refused it',))
PS C:\Users\aoighost\Desktop\motokodev>
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.