Update arvados backend for latest WES

Author: Peter Amstutz

setup.py

@@ -30,7 +30,7 @@
           'connexion',
           'bravado',
           'ruamel.yaml >= 0.12.4, < 0.15',
-          'cwl-runner'
+          'cwlref-runner'
         ],
       entry_points={
           'console_scripts': [ "wes-server=wes_service:main",

testdata/md5sum.cwl.json

@@ -1,10 +1,6 @@
 {
   "input_file": {
         "class": "File",
-        "path": "../../testdata/md5sum.input"
-    },
-    "output_file": {
-        "class": "File",
-        "path": "/tmp/md5sum.txt"
+        "path": "md5sum.input"
     }
-}
+}

wes_client/__init__.py

@@ -80,6 +80,11 @@ def main(argv=sys.argv[1:]):
         basedir = os.path.dirname(args.job_order)
 
         def fixpaths(d):
+            if isinstance(d, dict) and "path" in d:
+                local_path = os.path.normpath(
+                    os.path.join(os.getcwd(), basedir, d["path"]))
+                del d["path"]
+                d["location"] = urllib.pathname2url(local_path)
             if isinstance(d, dict) and "location" in d:
                 if ":" not in d["location"]:
                     local_path = os.path.normpath(

wes_service/arvados_wes.py

@@ -1,4 +1,5 @@
 import arvados
+import arvados.util
 import arvados.collection
 import os
 import connexion
@@ -17,11 +18,11 @@ def get_api():
 
 
 statemap = {
-    "Queued": "Queued",
-    "Locked": "Initializing",
-    "Running": "Running",
-    "Complete": "Complete",
-    "Cancelled": "Canceled"
+    "Queued": "QUEUED",
+    "Locked": "INITIALIZING",
+    "Running": "RUNNING",
+    "Complete": "COMPLETE",
+    "Cancelled": "CANCELED"
 }
 
 
@@ -31,8 +32,8 @@ def GetServiceInfo(self):
             "workflow_type_versions": {
                 "CWL": ["v1.0"]
             },
-            "supported_wes_versions": "0.1.0",
-            "supported_filesystem_protocols": ["file"],
+            "supported_wes_versions": "0.2.1",
+            "supported_filesystem_protocols": ["file", "http", "https", "keep"],
             "engine_versions": "cwl-runner",
             "system_state_counts": {},
             "key_values": {}
@@ -41,19 +42,21 @@ def GetServiceInfo(self):
     def ListWorkflows(self):
         api = get_api()
 
-        requests = api.container_requests().list(
-            filters=[["requesting_container_uuid", "=", None]],
-            select=["uuid", "command", "container_uuid"]).execute()
-        containers = api.containers().list(filters=[["uuid", "in", [w["container_uuid"] for w in requests["items"]]]],  # NOQA
-                                           select=["uuid", "state"]).execute()
+        requests = arvados.util.list_all(api.container_requests().list,
+                                         filters=[["requesting_container_uuid", "=", None],
+                                                  ["container_uuid", "!=", None]],
+                                         select=["uuid", "command", "container_uuid"])
+        containers = arvados.util.list_all(api.containers().list,
+                                           filters=[["uuid", "in", [w["container_uuid"] for w in requests]]],
+                                           select=["uuid", "state"])
 
-        uuidmap = {c["uuid"]: statemap[c["state"]] for c in containers["items"]}  # NOQA
+        uuidmap = {c["uuid"]: statemap[c["state"]] for c in containers}
 
         return {
             "workflows": [{"workflow_id": cr["uuid"],
-                           "state": uuidmap[cr["container_uuid"]]}
-                          for cr in requests["items"]
-                          if cr["command"][0] == "arvados-cwl-runner"],
+                           "state": uuidmap.get(cr["container_uuid"])}
+                          for cr in requests
+                          if cr["command"] and cr["command"][0] == "arvados-cwl-runner"],
             "next_page_token": ""
         }
 
@@ -114,7 +117,7 @@ def keepref(d):
             "outputs": outputobj
         }
         if container["exit_code"] is not None:
-            r["workflow_log"]["exitCode"] = container["exit_code"]
+            r["workflow_log"]["exit_code"] = container["exit_code"]
         return r
 
     def CancelJob(self, workflow_id):  # NOQA