Expand server and client to have proper command line interface.

Author: Peter Amstutz

setup.py

@@ -19,7 +19,9 @@
       url="https://github.com/common-workflow-language/cwltool-service",
       download_url="https://github.com/common-workflow-language/cwltool-service",
       license='Apache 2.0',
-      py_modules=["cwl_runner_wes"],
+      packages=["wes_service"],
+      package_data={'wes_service': ['swagger/proto/workflow_execution.swagger.json']},
+      include_package_data=True,
       install_requires=[
           'connexion',
           'bravado'

wes_client.py

@@ -6,28 +6,64 @@
 import time
 import pprint
 import sys
+import os
+import argparse
+import logging
 
-f = open("swagger/proto/workflow_execution.swagger.json")
-client = SwaggerClient.from_spec(json.load(f), origin_url="http://localhost:8080")
+def main(argv=sys.argv[1:]):
 
-with open(sys.argv[2]) as f:
-    input = json.load(f)
+    parser = argparse.ArgumentParser(description='Workflow Execution Service')
+    parser.add_argument("--host", type=str, default=os.environ.get("WES_API_HOST"))
+    parser.add_argument("--auth", type=str, default=os.environ.get("WES_API_TOKEN"))
+    parser.add_argument("--proto", type=str, default="https")
+    parser.add_argument("--quiet", action="store_true", default=False)
+    parser.add_argument("workflow_url", type=str)
+    parser.add_argument("job_order", type=str)
+    args = parser.parse_args(argv)
 
-r = client.WorkflowExecutionService.RunWorkflow(body={
-    "workflow_url": sys.argv[1],
-    "workflow_params": input,
-    "workflow_type": "CWL",
-    "workflow_type_version": "v1.0"}).result()
+    http_client = RequestsClient()
+    http_client.set_api_key(
+        args.host, args.auth,
+        param_name='Authorization', param_in='header')
+    client = SwaggerClient.from_url("%s://%s/swagger.json" % (args.proto, args.host), http_client=http_client)
 
-sys.stderr.write(r.workflow_id+"\n")
+    with open(args.job_order) as f:
+        input = json.load(f)
+
+    workflow_url = args.workflow_url
+    if not workflow_url.startswith("/") or ":" in workflow_url:
+        workflow_url = os.path.abspath(workflow_url)
+
+    if args.quiet:
+        logging.basicConfig(level=logging.WARNING)
+    else:
+        logging.basicConfig(level=logging.INFO)
+
+    r = client.WorkflowExecutionService.RunWorkflow(body={
+        "workflow_url": workflow_url,
+        "workflow_params": input,
+        "workflow_type": "CWL",
+        "workflow_type_version": "v1.0"}).result()
+
+    logging.info("Workflow id is %s", r.workflow_id)
 
-r = client.WorkflowExecutionService.GetWorkflowStatus(workflow_id=r.workflow_id).result()
-while r.state == "Running":
-    time.sleep(1)
     r = client.WorkflowExecutionService.GetWorkflowStatus(workflow_id=r.workflow_id).result()
+    while r.state == "Running":
+        time.sleep(1)
+        r = client.WorkflowExecutionService.GetWorkflowStatus(workflow_id=r.workflow_id).result()
+
+    logging.info("State is %s", r.state)
+
+    s = client.WorkflowExecutionService.GetWorkflowLog(workflow_id=r.workflow_id).result()
+    logging.info(s.workflow_log.stderr)
+
+    d = {k: s.outputs[k] for k in s.outputs if k != "fields"}
+    json.dump(d, sys.stdout, indent=4)
 
-s = client.WorkflowExecutionService.GetWorkflowLog(workflow_id=r.workflow_id).result()
-sys.stderr.write(s.workflow_log.stderr+"\n")
+    if r.state == "Complete":
+        return 0
+    else:
+        return 1
 
-d = {k: s.outputs[k] for k in s.outputs if k != "fields"}
-json.dump(d, sys.stdout, indent=4)
+if __name__ == "__main__":
+    sys.exit(main(sys.argv[1:]))

wes_service.py renamed to wes_service/__init__.py

@@ -10,6 +10,8 @@
 import json
 import urllib
 
+from pkg_resources import resource_stream
+
 class Workflow(object):
     def __init__(self, workflow_id):
         super(Workflow, self).__init__()
@@ -92,6 +94,7 @@ def getlog(self):
         with open(os.path.join(self.workdir, "stderr"), "r") as f:
             stderr = f.read()
 
+        outputobj = {}
         if state == "Complete":
             with open(os.path.join(self.workdir, "cwl.output.json"), "r") as outputtemp:
                 outputobj = json.load(outputtemp)
@@ -163,11 +166,12 @@ def GetWorkflowStatus(workflow_id):
     return job.getstatus()
 
 def main():
-    app = connexion.App(__name__, specification_dir='swagger/')
+    app = connexion.App(__name__)
     def rs(x):
         return utils.get_function_from_name("wes_service." + x)
 
-    app.add_api('proto/workflow_execution.swagger.json', resolver=Resolver(rs))
+    res = resource_stream(__name__, 'swagger/proto/workflow_execution.swagger.json')
+    app.add_api(json.load(res), resolver=Resolver(rs))
 
     app.run(port=8080)
 

swagger/proto/workflow_execution.swagger.json renamed to wes_service/swagger/proto/workflow_execution.swagger.json

@@ -1,5 +1,6 @@
 {
-  "swagger": "2.0",
+    "swagger": "2.0",
+    "basePath": "/",
   "info": {
     "title": "workflow_execution.proto",
     "version": "version not set"