Multipart upload (#22)

* Improvements to arvados_wes around logs and paging.

* Add page-size to wes-client

* Proof of concept using multipart upload.

* Set cwd to tempdir, clean up afterwards.

* Update to latest multipart proposal.

* Fix interface

Author: tetron

wes_client/__init__.py

@@ -9,18 +9,20 @@
 import argparse
 import logging
 import schema_salad.ref_resolver
+import requests
 from wes_service.util import visit
 from bravado.client import SwaggerClient
 from bravado.requests_client import RequestsClient
 
-
 def main(argv=sys.argv[1:]):
     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_AUTH"))
     parser.add_argument("--proto", type=str, default=os.environ.get("WES_API_PROTO", "https"))
     parser.add_argument("--quiet", action="store_true", default=False)
     parser.add_argument("--outdir", type=str)
+    parser.add_argument("--page", type=str, default=None)
+    parser.add_argument("--page-size", type=int, default=None)
 
     exgroup = parser.add_mutually_exclusive_group()
     exgroup.add_argument("--run", action="store_true", default=False)
@@ -54,7 +56,7 @@ def main(argv=sys.argv[1:]):
         http_client=http_client, config={'use_models': False})
 
     if args.list:
-        response = client.WorkflowExecutionService.ListWorkflows()
+        response = client.WorkflowExecutionService.ListWorkflows(page_token=args.page, page_size=args.page_size)
         json.dump(response.result(), sys.stdout, indent=4)
         return 0
 
@@ -98,10 +100,10 @@ def fixpaths(d):
                 if loc.startswith("http:") or loc.startswith("https:"):
                     logging.error("Directory inputs not supported with http references")
                     exit(33)
-            if not (loc.startswith("http:") or loc.startswith("https:")
-                    or args.job_order.startswith("http:") or args.job_order.startswith("https:")):
-                logging.error("Upload local files not supported, must use http: or https: references.")
-                exit(33)
+            # if not (loc.startswith("http:") or loc.startswith("https:")
+            #         or args.job_order.startswith("http:") or args.job_order.startswith("https:")):
+            #     logging.error("Upload local files not supported, must use http: or https: references.")
+            #     exit(33)
 
     visit(input, fixpaths)
 
@@ -114,19 +116,37 @@ def fixpaths(d):
     else:
         logging.basicConfig(level=logging.INFO)
 
-    body = {
-        "workflow_params": input,
-        "workflow_type": "CWL",
-        "workflow_type_version": "v1.0"
-    }
+    parts = [
+        ("workflow_params", json.dumps(input)),
+        ("workflow_type", "CWL"),
+        ("workflow_type_version", "v1.0")
+    ]
 
     if workflow_url.startswith("file://"):
-        with open(workflow_url[7:], "r") as f:
-            body["workflow_descriptor"] = f.read()
+        # with open(workflow_url[7:], "rb") as f:
+        #     body["workflow_descriptor"] = f.read()
+        rootdir = os.path.dirname(workflow_url[7:])
+        dirpath = rootdir
+        #for dirpath, dirnames, filenames in os.walk(rootdir):
+        for f in os.listdir(rootdir):
+            if f.startswith("."):
+                continue
+            fn = os.path.join(dirpath, f)
+            if os.path.isfile(fn):
+                parts.append(('workflow_descriptor', (fn[len(rootdir)+1:], open(fn, "rb"))))
+        parts.append(("workflow_url", os.path.basename(workflow_url[7:])))
     else:
-        body["workflow_url"] = workflow_url
+        parts.append(("workflow_url", workflow_url))
+
+    postresult = http_client.session.post("%s://%s/ga4gh/wes/v1/workflows" % (args.proto, args.host),
+                                          files=parts,
+                                          headers={"Authorization": args.auth})
+
+    r = json.loads(postresult.text)
 
-    r = client.WorkflowExecutionService.RunWorkflow(body=body).result()
+    if postresult.status_code != 200:
+        logging.error("%s", r)
+        exit(1)
 
     if args.wait:
         logging.info("Workflow id is %s", r["workflow_id"])
@@ -137,15 +157,17 @@ def fixpaths(d):
     r = client.WorkflowExecutionService.GetWorkflowStatus(
         workflow_id=r["workflow_id"]).result()
     while r["state"] in ("QUEUED", "INITIALIZING", "RUNNING"):
-        time.sleep(1)
+        time.sleep(8)
         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("Workflow log:\n"+s["workflow_log"]["stderr"])
+    logging.info("%s", s["workflow_log"]["stderr"])
+    logs = requests.get(s["workflow_log"]["stderr"], headers={"Authorization": args.auth}).text
+    logging.info("Workflow log:\n"+logs)
 
     if "fields" in s["outputs"] and s["outputs"]["fields"] is None:
         del s["outputs"]["fields"]

wes_service/__init__.py

@@ -28,10 +28,11 @@ def setup(args=None):
         logging.info("  %s: %s", n, getattr(args, n))
 
     app = connexion.App(__name__)
-    backend = utils.get_function_from_name(args.backend + ".create_backend")(args.opt)
+    backend = utils.get_function_from_name(
+        args.backend + ".create_backend")(app, args.opt)
 
     def rs(x):
-        return getattr(backend, x)
+        return getattr(backend, x.split('.')[-1])
 
     app.add_api(
         'openapi/workflow_execution_service.swagger.yaml',