Update to connexion 3

Co-authored-by: Michael R. Crusoe <[email protected]>

Author: stxue1

mypy-requirements.txt

@@ -3,3 +3,4 @@ types-PyYAML
 types-requests
 types-setuptools
 arvados-cwl-runner
+flask

pyproject.toml

@@ -23,7 +23,7 @@ classifiers = [
 ]
 requires-python = ">=3.9"
 dependencies = [
-    "connexion[swagger-ui] >= 2.0.2, < 3",
+    "connexion[swagger-ui] >= 3, < 4",
     "ruamel.yaml >= 0.15.78",
     "schema-salad",
 ]

wes_service/arvados_wes.py

@@ -298,7 +298,7 @@ def RunWorkflow(
         )
 
         try:
-            tempdir, body = self.collect_attachments(cr["uuid"])
+            tempdir, body = self.collect_attachments(args, cr["uuid"])
 
             workflow_engine_parameters = cast(
                 dict[str, Any], body.get("workflow_engine_parameters", {})

wes_service/cwl_runner.py

@@ -198,7 +198,7 @@ def ListRuns(
 
     def RunWorkflow(self, **args: str) -> dict[str, str]:
         """Submit the workflow run request."""
-        tempdir, body = self.collect_attachments()
+        tempdir, body = self.collect_attachments(args)
 
         run_id = uuid.uuid4().hex
         job = Workflow(run_id)

wes_service/toil_wes.py

@@ -1,3 +1,6 @@
+"""Toil backed for the WES service."""
+
+import errno
 import json
 import logging
 import os
@@ -294,18 +297,36 @@ def getstate(self) -> tuple[str, int]:
                     logging.info("Workflow " + self.run_id + ": EXECUTOR_ERROR")
                     open(self.staterrorfile, "a").close()
                     return "EXECUTOR_ERROR", 255
+
+        # get the jobstore
+        with open(self.jobstorefile, "r") as f:
+            jobstore = f.read().rstrip()
         if (
             subprocess.run(  # nosec B603
                 [
                     shutil.which("toil") or "toil",
                     "status",
                     "--failIfNotComplete",
-                    self.jobstorefile,
+                    jobstore,
                 ]
             ).returncode
             == 0
         ):
-            completed = True
+            # Get the PID of the running process
+            with open(self.pidfile, "r") as f:
+                pid = int(f.read())
+            try:
+                os.kill(pid, 0)
+            except OSError as e:
+                if e.errno == errno.ESRCH:
+                    # Process is no longer running, could be completed
+                    completed = True
+                    # Reap zombie child processes in a non-blocking manner
+                    os.waitpid(pid, os.WNOHANG)
+                else:
+                    raise
+            # If no exception, process is still running
+            # We can't rely on toil status as the process may not have created the jobstore yet
         if completed:
             logging.info("Workflow " + self.run_id + ": COMPLETE")
             open(self.statcompletefile, "a").close()
@@ -354,9 +375,9 @@ def ListRuns(
         workflows = [{"run_id": w.run_id, "state": w.getstate()[0]} for w in wf]  # NOQA
         return {"workflows": workflows, "next_page_token": ""}
 
-    def RunWorkflow(self) -> dict[str, str]:
+    def RunWorkflow(self, **args: str) -> dict[str, str]:
         """Submit the workflow run request."""
-        tempdir, body = self.collect_attachments()
+        tempdir, body = self.collect_attachments(args)
 
         run_id = uuid.uuid4().hex
         job = ToilWorkflow(run_id)

wes_service/util.py

@@ -4,7 +4,6 @@
 import tempfile
 from typing import Any, Callable, Optional
 
-import connexion  # type: ignore[import-untyped]
 from werkzeug.utils import secure_filename
 
 
@@ -49,52 +48,37 @@ def log_for_run(self, run_id: Optional[str], message: str) -> None:
         logging.info("Workflow %s: %s", run_id, message)
 
     def collect_attachments(
-        self, run_id: Optional[str] = None
+        self, args: dict[str, Any], run_id: Optional[str] = None
     ) -> tuple[str, dict[str, str]]:
         """Stage all attachments to a temporary directory."""
         tempdir = tempfile.mkdtemp()
         body: dict[str, str] = {}
         has_attachments = False
-        for k, ls in connexion.request.files.lists():
-            try:
-                for v in ls:
-                    if k == "workflow_attachment":
-                        sp = v.filename.split("/")
-                        fn = []
-                        for p in sp:
-                            if p not in ("", ".", ".."):
-                                fn.append(secure_filename(p))
-                        dest = os.path.join(tempdir, *fn)
-                        if not os.path.isdir(os.path.dirname(dest)):
-                            os.makedirs(os.path.dirname(dest))
-                        self.log_for_run(
-                            run_id,
-                            f"Staging attachment {v.filename!r} to {dest!r}",
-                        )
-                        v.save(dest)
-                        has_attachments = True
-                        body[k] = (
-                            "file://%s" % tempdir
-                        )  # Reference to temp working dir.
-                    elif k in ("workflow_params", "tags", "workflow_engine_parameters"):
-                        content = v.read()
-                        body[k] = json.loads(content.decode("utf-8"))
-                    else:
-                        body[k] = v.read().decode()
-            except Exception as e:
-                raise ValueError(f"Error reading parameter {k!r}: {e}") from e
-        for k, ls in connexion.request.form.lists():
-            try:
-                for v in ls:
-                    if not v:
-                        continue
-                    if k in ("workflow_params", "tags", "workflow_engine_parameters"):
-                        body[k] = json.loads(v)
-                    else:
-                        body[k] = v
-            except Exception as e:
-                raise ValueError(f"Error reading parameter {k!r}: {e}") from e
-
+        for k, v in args.items():
+            if k == "workflow_attachment":
+                for file in v or []:
+                    sp = file.filename.split("/")
+                    fn = []
+                    for p in sp:
+                        if p not in ("", ".", ".."):
+                            fn.append(secure_filename(p))
+                    dest = os.path.join(tempdir, *fn)
+                    if not os.path.isdir(os.path.dirname(dest)):
+                        os.makedirs(os.path.dirname(dest))
+                    self.log_for_run(
+                        run_id,
+                        f"Staging attachment {file.filename!r} to {dest!r}",
+                    )
+                    file.save(dest)
+                    has_attachments = True
+                    body["workflow_attachment"] = (
+                        "file://%s" % tempdir
+                    )  # Reference to temp working dir.
+            elif k in ("workflow_params", "tags", "workflow_engine_parameters"):
+                if v is not None:
+                    body[k] = json.loads(v)
+            else:
+                body[k] = v
         if "workflow_url" in body:
             if ":" not in body["workflow_url"]:
                 if not has_attachments:

wes_service/wes_service_main.py

@@ -63,7 +63,6 @@ def get_parser() -> argparse.ArgumentParser:
         help="Example: '--opt runner=cwltoil --opt extra=--logLevel=CRITICAL' "
         "or '--opt extra=--workDir=/'.  Accepts multiple values.",
     )
-    parser.add_argument("--debug", action="store_true", default=False)
     parser.add_argument("--version", action="store_true", default=False)
     return parser
 
@@ -78,7 +77,7 @@ def main(argv: list[str] = sys.argv[1:]) -> None:
 
     app = setup(args)
 
-    app.run(port=args.port, debug=args.debug)
+    app.run(port=args.port)
 
 
 if __name__ == "__main__":