We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
We're getting the new Werkzeug without asking for it
CI for the merge https://github.com/DataBiosphere/toil/commit/0e256d63cb974a87b8f6b807bf7d23bc9a12fb76 failed at the lint stage, because the merge commit ends up installing a different Werkzeug than the PR's test run did, and the new one has type hints, which upsets MyPy because we now have an unused ignore.

This is because the `connexion` devs finally got access to the `connexion` PyPI package again, and published the current release there. So we started picking up connexion 2.10 instead of 2.5, which is now compatible with Flask 2. So we started installing Flask 2 and Werkzeug 2.

If we're going to import out of Werkzeug, we need to depend on a particular major version of it, so it can't be changed put from under us by pip.

┆Issue is synchronized with this [Jira Task](https://ucsc-cgl.atlassian.net/browse/TOIL-1130)
┆friendlyId: TOIL-1130


--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `src/toil/server/wes/abstract_backend.py`
Content:
```
1 # Modified from: https://github.com/common-workflow-language/workflow-service
2 import functools
3 import json
4 import os
5 import logging
6 import tempfile
7 from abc import abstractmethod
8 from typing import Optional, List, Dict, Any, Tuple, Callable
9 from urllib.parse import urldefrag
10 
11 import connexion  # type: ignore
12 from werkzeug.utils import secure_filename  # type: ignore
13 
14 logger = logging.getLogger(__name__)
15 
16 
17 class VersionNotImplementedException(Exception):
18     """
19     Raised when the requested workflow version is not implemented.
20     """
21     def __init__(self,
22                  wf_type: str, version: Optional[str] = None, supported_versions: Optional[List[str]] = None) -> None:
23         if version:
24             message = ("workflow_type '{}' requires 'workflow_type_version' to be one of '{}'.  "
25                        "Got '{}' instead.".format(wf_type, str(supported_versions), version))
26         else:
27             message = f"workflow_type '{wf_type}' is not supported."
28 
29         super(VersionNotImplementedException, self).__init__(message)
30 
31 
32 class MalformedRequestException(Exception):
33     """
34     Raised when the request is malformed.
35     """
36     def __init__(self, message: str) -> None:
37         super(MalformedRequestException, self).__init__(message)
38 
39 
40 class WorkflowNotFoundException(Exception):
41     """
42     Raised when the requested run ID is not found.
43     """
44     def __init__(self) -> None:
45         super(WorkflowNotFoundException, self).__init__("The requested workflow run wasn't found.")
46 
47 
48 class WorkflowConflictException(Exception):
49     """
50     Raised when the requested workflow is not in the expected state.
51     """
52     def __init__(self, run_id: str):
53         super(WorkflowConflictException, self).__init__(f"Workflow {run_id} exists when it shouldn't.")
54 
55 
56 class OperationForbidden(Exception):
57     """
58     Raised when the request is forbidden.
59     """
60     def __init__(self, message: str) -> None:
61         super(OperationForbidden, self).__init__(message)
62 
63 
64 class WorkflowExecutionException(Exception):
65     """
66     Raised when an internal error occurred during the execution of the workflow.
67     """
68     def __init__(self, message: str) -> None:
69         super(WorkflowExecutionException, self).__init__(message)
70 
71 
72 def handle_errors(func: Callable[..., Any]) -> Callable[..., Any]:
73     """
74     This decorator catches errors from the wrapped function and returns a JSON
75     formatted error message with the appropriate status code defined by the
76     GA4GH WES spec.
77     """
78 
79     def error(msg: Any, code: int = 500) -> Tuple[Dict[str, Any], int]:
80         logger.warning(f"Exception raised when calling '{func.__name__}()':", exc_info=True)
81         return {"msg": str(msg), "status_code": code}, code
82 
83     @functools.wraps(func)
84     def wrapper(*args: Any, **kwargs: Any) -> Any:
85         try:
86             return func(*args, **kwargs)
87         except MalformedRequestException as e:
88             return error(e, code=400)
89         except VersionNotImplementedException as e:
90             return error(e, code=400)
91         except OperationForbidden as e:
92             return error(e, code=403)
93         except (FileNotFoundError, WorkflowNotFoundException) as e:
94             return error(e, code=404)
95         except WorkflowConflictException as e:
96             return error(e, code=400)
97         except WorkflowExecutionException as e:
98             return error(e, code=500)
99         except Exception as e:
100             return error(e, code=500)
101 
102     return wrapper
103 
104 
105 class WESBackend:
106     """
107     A class to represent a GA4GH Workflow Execution Service (WES) API backend.
108     Intended to be inherited. Subclasses should implement all abstract methods
109     to handle user requests when they hit different endpoints.
110     """
111 
112     def __init__(self, options: List[str]):
113         """
114         :param options: A list of default engine options to use when executing
115                         a workflow.  Example options:
116                         ["--logLevel=CRITICAL","--workDir=/path/to/dir",
117                             "--tag=Name=default", "--tag=Owner=shared", ...]
118         """
119         self.options = options or []
120 
121     def resolve_operation_id(self, operation_id: str) -> Any:
122         """
123         Map an operationId defined in the OpenAPI or swagger yaml file to a
124         function.
125 
126         :param operation_id: The operation ID defined in the specification.
127         :returns: A function that should be called when the given endpoint is
128                   reached.
129         """
130         return getattr(self, operation_id.split(".")[-1])
131 
132     @abstractmethod
133     def get_service_info(self) -> Dict[str, Any]:
134         """
135         Get information about the Workflow Execution Service.
136 
137         GET /service-info
138         """
139         raise NotImplementedError
140 
141     @abstractmethod
142     def list_runs(self, page_size: Optional[int] = None, page_token: Optional[str] = None) -> Dict[str, Any]:
143         """
144         List the workflow runs.
145 
146         GET /runs
147         """
148         raise NotImplementedError
149 
150     @abstractmethod
151     def run_workflow(self) -> Dict[str, str]:
152         """
153         Run a workflow. This endpoint creates a new workflow run and returns
154         a `RunId` to monitor its progress.
155 
156         POST /runs
157         """
158         raise NotImplementedError
159 
160     @abstractmethod
161     def get_run_log(self, run_id: str) -> Dict[str, Any]:
162         """
163         Get detailed info about a workflow run.
164 
165         GET /runs/{run_id}
166         """
167         raise NotImplementedError
168 
169     @abstractmethod
170     def cancel_run(self, run_id: str) -> Dict[str, str]:
171         """
172         Cancel a running workflow.
173 
174         POST /runs/{run_id}/cancel
175         """
176         raise NotImplementedError
177 
178     @abstractmethod
179     def get_run_status(self, run_id: str) -> Dict[str, str]:
180         """
181         Get quick status info about a workflow run, returning a simple result
182         with the overall state of the workflow run.
183 
184         GET /runs/{run_id}/status
185         """
186         raise NotImplementedError
187 
188     @staticmethod
189     def log_for_run(run_id: Optional[str], message: str) -> None:
190         if run_id:
191             logging.info("Workflow %s: %s", run_id, message)
192         else:
193             logging.info(message)
194 
195     @staticmethod
196     def secure_path(path: str) -> str:
197         return os.path.join(*[str(secure_filename(p)) for p in path.split("/") if p not in ("", ".", "..")])
198 
199     def collect_attachments(self, run_id: Optional[str], temp_dir: Optional[str]) -> Tuple[str, Dict[str, Any]]:
200         """
201         Collect attachments from the current request by staging uploaded files
202         to temp_dir, and return the temp_dir and parsed body of the request.
203 
204         :param run_id: The run ID for logging.
205         :param temp_dir: The directory where uploaded files should be staged.
206                          If None, a temporary directory is created.
207         """
208         if not temp_dir:
209             temp_dir = tempfile.mkdtemp()
210         body = {}
211         has_attachments = False
212         for key, ls in connexion.request.files.lists():
213             try:
214                 for value in ls:
215                     # uploaded files that are required to execute the workflow
216                     if key == "workflow_attachment":
217                         # guard against maliciously constructed filenames
218                         dest = os.path.join(temp_dir, self.secure_path(value.filename))
219                         if not os.path.isdir(os.path.dirname(dest)):
220                             os.makedirs(os.path.dirname(dest))
221                         self.log_for_run(run_id, f"Staging attachment '{value.filename}' to '{dest}'")
222                         value.save(dest)
223                         has_attachments = True
224                         body[key] = f"file://{temp_dir}"  # Reference to temp working dir.
225 
226                     elif key in ("workflow_params", "tags", "workflow_engine_parameters"):
227                         content = value.read()
228                         body[key] = json.loads(content.decode("utf-8"))
229                     else:
230                         body[key] = value.read().decode()
231             except Exception as e:
232                 raise MalformedRequestException(f"Error reading parameter '{key}': {e}")
233 
234         for key, ls in connexion.request.form.lists():
235             try:
236                 for value in ls:
237                     if not value:
238                         continue
239                     if key in ("workflow_params", "tags", "workflow_engine_parameters"):
240                         body[key] = json.loads(value)
241                     else:
242                         body[key] = value
243             except Exception as e:
244                 raise MalformedRequestException(f"Error reading parameter '{key}': {e}")
245 
246         if "workflow_url" in body:
247             url, ref = urldefrag(body["workflow_url"])
248             if ":" not in url:
249                 if not has_attachments:
250                     raise MalformedRequestException("Relative 'workflow_url' but missing 'workflow_attachment'")
251                 body["workflow_url"] = self.secure_path(url)  # keep this relative
252                 if ref:
253                     # append "#ref" after the url
254                     body["workflow_url"] += "#" + self.secure_path(ref)
255             self.log_for_run(run_id, "Using workflow_url '%s'" % body.get("workflow_url"))
256         else:
257             raise MalformedRequestException("Missing 'workflow_url' in submission")
258         if "workflow_params" not in body:
259             raise MalformedRequestException("Missing 'workflow_params' in submission")
260 
261         return temp_dir, body
262
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/src/toil/server/wes/abstract_backend.py b/src/toil/server/wes/abstract_backend.py
--- a/src/toil/server/wes/abstract_backend.py
+++ b/src/toil/server/wes/abstract_backend.py
@@ -9,7 +9,7 @@
 from urllib.parse import urldefrag
 
 import connexion  # type: ignore
-from werkzeug.utils import secure_filename  # type: ignore
+from werkzeug.utils import secure_filename
 
 logger = logging.getLogger(__name__)

{"golden_diff": "diff --git a/src/toil/server/wes/abstract_backend.py b/src/toil/server/wes/abstract_backend.py\n--- a/src/toil/server/wes/abstract_backend.py\n+++ b/src/toil/server/wes/abstract_backend.py\n@@ -9,7 +9,7 @@\n from urllib.parse import urldefrag\n \n import connexion  # type: ignore\n-from werkzeug.utils import secure_filename  # type: ignore\n+from werkzeug.utils import secure_filename\n \n logger = logging.getLogger(__name__)\n", "issue": "We're getting the new Werkzeug without asking for it\nCI for the merge https://github.com/DataBiosphere/toil/commit/0e256d63cb974a87b8f6b807bf7d23bc9a12fb76 failed at the lint stage, because the merge commit ends up installing a different Werkzeug than the PR's test run did, and the new one has type hints, which upsets MyPy because we now have an unused ignore.\n\nThis is because the `connexion` devs finally got access to the `connexion` PyPI package again, and published the current release there. So we started picking up connexion 2.10 instead of 2.5, which is now compatible with Flask 2. So we started installing Flask 2 and Werkzeug 2.\n\nIf we're going to import out of Werkzeug, we need to depend on a particular major version of it, so it can't be changed put from under us by pip.\n\n\u2506Issue is synchronized with this [Jira Task](https://ucsc-cgl.atlassian.net/browse/TOIL-1130)\n\u2506friendlyId: TOIL-1130\n\n", "before_files": [{"content": "# Modified from: https://github.com/common-workflow-language/workflow-service\nimport functools\nimport json\nimport os\nimport logging\nimport tempfile\nfrom abc import abstractmethod\nfrom typing import Optional, List, Dict, Any, Tuple, Callable\nfrom urllib.parse import urldefrag\n\nimport connexion  # type: ignore\nfrom werkzeug.utils import secure_filename  # type: ignore\n\nlogger = logging.getLogger(__name__)\n\n\nclass VersionNotImplementedException(Exception):\n    \"\"\"\n    Raised when the requested workflow version is not implemented.\n    \"\"\"\n    def __init__(self,\n                 wf_type: str, version: Optional[str] = None, supported_versions: Optional[List[str]] = None) -> None:\n        if version:\n            message = (\"workflow_type '{}' requires 'workflow_type_version' to be one of '{}'.  \"\n                       \"Got '{}' instead.\".format(wf_type, str(supported_versions), version))\n        else:\n            message = f\"workflow_type '{wf_type}' is not supported.\"\n\n        super(VersionNotImplementedException, self).__init__(message)\n\n\nclass MalformedRequestException(Exception):\n    \"\"\"\n    Raised when the request is malformed.\n    \"\"\"\n    def __init__(self, message: str) -> None:\n        super(MalformedRequestException, self).__init__(message)\n\n\nclass WorkflowNotFoundException(Exception):\n    \"\"\"\n    Raised when the requested run ID is not found.\n    \"\"\"\n    def __init__(self) -> None:\n        super(WorkflowNotFoundException, self).__init__(\"The requested workflow run wasn't found.\")\n\n\nclass WorkflowConflictException(Exception):\n    \"\"\"\n    Raised when the requested workflow is not in the expected state.\n    \"\"\"\n    def __init__(self, run_id: str):\n        super(WorkflowConflictException, self).__init__(f\"Workflow {run_id} exists when it shouldn't.\")\n\n\nclass OperationForbidden(Exception):\n    \"\"\"\n    Raised when the request is forbidden.\n    \"\"\"\n    def __init__(self, message: str) -> None:\n        super(OperationForbidden, self).__init__(message)\n\n\nclass WorkflowExecutionException(Exception):\n    \"\"\"\n    Raised when an internal error occurred during the execution of the workflow.\n    \"\"\"\n    def __init__(self, message: str) -> None:\n        super(WorkflowExecutionException, self).__init__(message)\n\n\ndef handle_errors(func: Callable[..., Any]) -> Callable[..., Any]:\n    \"\"\"\n    This decorator catches errors from the wrapped function and returns a JSON\n    formatted error message with the appropriate status code defined by the\n    GA4GH WES spec.\n    \"\"\"\n\n    def error(msg: Any, code: int = 500) -> Tuple[Dict[str, Any], int]:\n        logger.warning(f\"Exception raised when calling '{func.__name__}()':\", exc_info=True)\n        return {\"msg\": str(msg), \"status_code\": code}, code\n\n    @functools.wraps(func)\n    def wrapper(*args: Any, **kwargs: Any) -> Any:\n        try:\n            return func(*args, **kwargs)\n        except MalformedRequestException as e:\n            return error(e, code=400)\n        except VersionNotImplementedException as e:\n            return error(e, code=400)\n        except OperationForbidden as e:\n            return error(e, code=403)\n        except (FileNotFoundError, WorkflowNotFoundException) as e:\n            return error(e, code=404)\n        except WorkflowConflictException as e:\n            return error(e, code=400)\n        except WorkflowExecutionException as e:\n            return error(e, code=500)\n        except Exception as e:\n            return error(e, code=500)\n\n    return wrapper\n\n\nclass WESBackend:\n    \"\"\"\n    A class to represent a GA4GH Workflow Execution Service (WES) API backend.\n    Intended to be inherited. Subclasses should implement all abstract methods\n    to handle user requests when they hit different endpoints.\n    \"\"\"\n\n    def __init__(self, options: List[str]):\n        \"\"\"\n        :param options: A list of default engine options to use when executing\n                        a workflow.  Example options:\n                        [\"--logLevel=CRITICAL\",\"--workDir=/path/to/dir\",\n                            \"--tag=Name=default\", \"--tag=Owner=shared\", ...]\n        \"\"\"\n        self.options = options or []\n\n    def resolve_operation_id(self, operation_id: str) -> Any:\n        \"\"\"\n        Map an operationId defined in the OpenAPI or swagger yaml file to a\n        function.\n\n        :param operation_id: The operation ID defined in the specification.\n        :returns: A function that should be called when the given endpoint is\n                  reached.\n        \"\"\"\n        return getattr(self, operation_id.split(\".\")[-1])\n\n    @abstractmethod\n    def get_service_info(self) -> Dict[str, Any]:\n        \"\"\"\n        Get information about the Workflow Execution Service.\n\n        GET /service-info\n        \"\"\"\n        raise NotImplementedError\n\n    @abstractmethod\n    def list_runs(self, page_size: Optional[int] = None, page_token: Optional[str] = None) -> Dict[str, Any]:\n        \"\"\"\n        List the workflow runs.\n\n        GET /runs\n        \"\"\"\n        raise NotImplementedError\n\n    @abstractmethod\n    def run_workflow(self) -> Dict[str, str]:\n        \"\"\"\n        Run a workflow. This endpoint creates a new workflow run and returns\n        a `RunId` to monitor its progress.\n\n        POST /runs\n        \"\"\"\n        raise NotImplementedError\n\n    @abstractmethod\n    def get_run_log(self, run_id: str) -> Dict[str, Any]:\n        \"\"\"\n        Get detailed info about a workflow run.\n\n        GET /runs/{run_id}\n        \"\"\"\n        raise NotImplementedError\n\n    @abstractmethod\n    def cancel_run(self, run_id: str) -> Dict[str, str]:\n        \"\"\"\n        Cancel a running workflow.\n\n        POST /runs/{run_id}/cancel\n        \"\"\"\n        raise NotImplementedError\n\n    @abstractmethod\n    def get_run_status(self, run_id: str) -> Dict[str, str]:\n        \"\"\"\n        Get quick status info about a workflow run, returning a simple result\n        with the overall state of the workflow run.\n\n        GET /runs/{run_id}/status\n        \"\"\"\n        raise NotImplementedError\n\n    @staticmethod\n    def log_for_run(run_id: Optional[str], message: str) -> None:\n        if run_id:\n            logging.info(\"Workflow %s: %s\", run_id, message)\n        else:\n            logging.info(message)\n\n    @staticmethod\n    def secure_path(path: str) -> str:\n        return os.path.join(*[str(secure_filename(p)) for p in path.split(\"/\") if p not in (\"\", \".\", \"..\")])\n\n    def collect_attachments(self, run_id: Optional[str], temp_dir: Optional[str]) -> Tuple[str, Dict[str, Any]]:\n        \"\"\"\n        Collect attachments from the current request by staging uploaded files\n        to temp_dir, and return the temp_dir and parsed body of the request.\n\n        :param run_id: The run ID for logging.\n        :param temp_dir: The directory where uploaded files should be staged.\n                         If None, a temporary directory is created.\n        \"\"\"\n        if not temp_dir:\n            temp_dir = tempfile.mkdtemp()\n        body = {}\n        has_attachments = False\n        for key, ls in connexion.request.files.lists():\n            try:\n                for value in ls:\n                    # uploaded files that are required to execute the workflow\n                    if key == \"workflow_attachment\":\n                        # guard against maliciously constructed filenames\n                        dest = os.path.join(temp_dir, self.secure_path(value.filename))\n                        if not os.path.isdir(os.path.dirname(dest)):\n                            os.makedirs(os.path.dirname(dest))\n                        self.log_for_run(run_id, f\"Staging attachment '{value.filename}' to '{dest}'\")\n                        value.save(dest)\n                        has_attachments = True\n                        body[key] = f\"file://{temp_dir}\"  # Reference to temp working dir.\n\n                    elif key in (\"workflow_params\", \"tags\", \"workflow_engine_parameters\"):\n                        content = value.read()\n                        body[key] = json.loads(content.decode(\"utf-8\"))\n                    else:\n                        body[key] = value.read().decode()\n            except Exception as e:\n                raise MalformedRequestException(f\"Error reading parameter '{key}': {e}\")\n\n        for key, ls in connexion.request.form.lists():\n            try:\n                for value in ls:\n                    if not value:\n                        continue\n                    if key in (\"workflow_params\", \"tags\", \"workflow_engine_parameters\"):\n                        body[key] = json.loads(value)\n                    else:\n                        body[key] = value\n            except Exception as e:\n                raise MalformedRequestException(f\"Error reading parameter '{key}': {e}\")\n\n        if \"workflow_url\" in body:\n            url, ref = urldefrag(body[\"workflow_url\"])\n            if \":\" not in url:\n                if not has_attachments:\n                    raise MalformedRequestException(\"Relative 'workflow_url' but missing 'workflow_attachment'\")\n                body[\"workflow_url\"] = self.secure_path(url)  # keep this relative\n                if ref:\n                    # append \"#ref\" after the url\n                    body[\"workflow_url\"] += \"#\" + self.secure_path(ref)\n            self.log_for_run(run_id, \"Using workflow_url '%s'\" % body.get(\"workflow_url\"))\n        else:\n            raise MalformedRequestException(\"Missing 'workflow_url' in submission\")\n        if \"workflow_params\" not in body:\n            raise MalformedRequestException(\"Missing 'workflow_params' in submission\")\n\n        return temp_dir, body\n", "path": "src/toil/server/wes/abstract_backend.py"}], "after_files": [{"content": "# Modified from: https://github.com/common-workflow-language/workflow-service\nimport functools\nimport json\nimport os\nimport logging\nimport tempfile\nfrom abc import abstractmethod\nfrom typing import Optional, List, Dict, Any, Tuple, Callable\nfrom urllib.parse import urldefrag\n\nimport connexion  # type: ignore\nfrom werkzeug.utils import secure_filename\n\nlogger = logging.getLogger(__name__)\n\n\nclass VersionNotImplementedException(Exception):\n    \"\"\"\n    Raised when the requested workflow version is not implemented.\n    \"\"\"\n    def __init__(self,\n                 wf_type: str, version: Optional[str] = None, supported_versions: Optional[List[str]] = None) -> None:\n        if version:\n            message = (\"workflow_type '{}' requires 'workflow_type_version' to be one of '{}'.  \"\n                       \"Got '{}' instead.\".format(wf_type, str(supported_versions), version))\n        else:\n            message = f\"workflow_type '{wf_type}' is not supported.\"\n\n        super(VersionNotImplementedException, self).__init__(message)\n\n\nclass MalformedRequestException(Exception):\n    \"\"\"\n    Raised when the request is malformed.\n    \"\"\"\n    def __init__(self, message: str) -> None:\n        super(MalformedRequestException, self).__init__(message)\n\n\nclass WorkflowNotFoundException(Exception):\n    \"\"\"\n    Raised when the requested run ID is not found.\n    \"\"\"\n    def __init__(self) -> None:\n        super(WorkflowNotFoundException, self).__init__(\"The requested workflow run wasn't found.\")\n\n\nclass WorkflowConflictException(Exception):\n    \"\"\"\n    Raised when the requested workflow is not in the expected state.\n    \"\"\"\n    def __init__(self, run_id: str):\n        super(WorkflowConflictException, self).__init__(f\"Workflow {run_id} exists when it shouldn't.\")\n\n\nclass OperationForbidden(Exception):\n    \"\"\"\n    Raised when the request is forbidden.\n    \"\"\"\n    def __init__(self, message: str) -> None:\n        super(OperationForbidden, self).__init__(message)\n\n\nclass WorkflowExecutionException(Exception):\n    \"\"\"\n    Raised when an internal error occurred during the execution of the workflow.\n    \"\"\"\n    def __init__(self, message: str) -> None:\n        super(WorkflowExecutionException, self).__init__(message)\n\n\ndef handle_errors(func: Callable[..., Any]) -> Callable[..., Any]:\n    \"\"\"\n    This decorator catches errors from the wrapped function and returns a JSON\n    formatted error message with the appropriate status code defined by the\n    GA4GH WES spec.\n    \"\"\"\n\n    def error(msg: Any, code: int = 500) -> Tuple[Dict[str, Any], int]:\n        logger.warning(f\"Exception raised when calling '{func.__name__}()':\", exc_info=True)\n        return {\"msg\": str(msg), \"status_code\": code}, code\n\n    @functools.wraps(func)\n    def wrapper(*args: Any, **kwargs: Any) -> Any:\n        try:\n            return func(*args, **kwargs)\n        except MalformedRequestException as e:\n            return error(e, code=400)\n        except VersionNotImplementedException as e:\n            return error(e, code=400)\n        except OperationForbidden as e:\n            return error(e, code=403)\n        except (FileNotFoundError, WorkflowNotFoundException) as e:\n            return error(e, code=404)\n        except WorkflowConflictException as e:\n            return error(e, code=400)\n        except WorkflowExecutionException as e:\n            return error(e, code=500)\n        except Exception as e:\n            return error(e, code=500)\n\n    return wrapper\n\n\nclass WESBackend:\n    \"\"\"\n    A class to represent a GA4GH Workflow Execution Service (WES) API backend.\n    Intended to be inherited. Subclasses should implement all abstract methods\n    to handle user requests when they hit different endpoints.\n    \"\"\"\n\n    def __init__(self, options: List[str]):\n        \"\"\"\n        :param options: A list of default engine options to use when executing\n                        a workflow.  Example options:\n                        [\"--logLevel=CRITICAL\",\"--workDir=/path/to/dir\",\n                            \"--tag=Name=default\", \"--tag=Owner=shared\", ...]\n        \"\"\"\n        self.options = options or []\n\n    def resolve_operation_id(self, operation_id: str) -> Any:\n        \"\"\"\n        Map an operationId defined in the OpenAPI or swagger yaml file to a\n        function.\n\n        :param operation_id: The operation ID defined in the specification.\n        :returns: A function that should be called when the given endpoint is\n                  reached.\n        \"\"\"\n        return getattr(self, operation_id.split(\".\")[-1])\n\n    @abstractmethod\n    def get_service_info(self) -> Dict[str, Any]:\n        \"\"\"\n        Get information about the Workflow Execution Service.\n\n        GET /service-info\n        \"\"\"\n        raise NotImplementedError\n\n    @abstractmethod\n    def list_runs(self, page_size: Optional[int] = None, page_token: Optional[str] = None) -> Dict[str, Any]:\n        \"\"\"\n        List the workflow runs.\n\n        GET /runs\n        \"\"\"\n        raise NotImplementedError\n\n    @abstractmethod\n    def run_workflow(self) -> Dict[str, str]:\n        \"\"\"\n        Run a workflow. This endpoint creates a new workflow run and returns\n        a `RunId` to monitor its progress.\n\n        POST /runs\n        \"\"\"\n        raise NotImplementedError\n\n    @abstractmethod\n    def get_run_log(self, run_id: str) -> Dict[str, Any]:\n        \"\"\"\n        Get detailed info about a workflow run.\n\n        GET /runs/{run_id}\n        \"\"\"\n        raise NotImplementedError\n\n    @abstractmethod\n    def cancel_run(self, run_id: str) -> Dict[str, str]:\n        \"\"\"\n        Cancel a running workflow.\n\n        POST /runs/{run_id}/cancel\n        \"\"\"\n        raise NotImplementedError\n\n    @abstractmethod\n    def get_run_status(self, run_id: str) -> Dict[str, str]:\n        \"\"\"\n        Get quick status info about a workflow run, returning a simple result\n        with the overall state of the workflow run.\n\n        GET /runs/{run_id}/status\n        \"\"\"\n        raise NotImplementedError\n\n    @staticmethod\n    def log_for_run(run_id: Optional[str], message: str) -> None:\n        if run_id:\n            logging.info(\"Workflow %s: %s\", run_id, message)\n        else:\n            logging.info(message)\n\n    @staticmethod\n    def secure_path(path: str) -> str:\n        return os.path.join(*[str(secure_filename(p)) for p in path.split(\"/\") if p not in (\"\", \".\", \"..\")])\n\n    def collect_attachments(self, run_id: Optional[str], temp_dir: Optional[str]) -> Tuple[str, Dict[str, Any]]:\n        \"\"\"\n        Collect attachments from the current request by staging uploaded files\n        to temp_dir, and return the temp_dir and parsed body of the request.\n\n        :param run_id: The run ID for logging.\n        :param temp_dir: The directory where uploaded files should be staged.\n                         If None, a temporary directory is created.\n        \"\"\"\n        if not temp_dir:\n            temp_dir = tempfile.mkdtemp()\n        body = {}\n        has_attachments = False\n        for key, ls in connexion.request.files.lists():\n            try:\n                for value in ls:\n                    # uploaded files that are required to execute the workflow\n                    if key == \"workflow_attachment\":\n                        # guard against maliciously constructed filenames\n                        dest = os.path.join(temp_dir, self.secure_path(value.filename))\n                        if not os.path.isdir(os.path.dirname(dest)):\n                            os.makedirs(os.path.dirname(dest))\n                        self.log_for_run(run_id, f\"Staging attachment '{value.filename}' to '{dest}'\")\n                        value.save(dest)\n                        has_attachments = True\n                        body[key] = f\"file://{temp_dir}\"  # Reference to temp working dir.\n\n                    elif key in (\"workflow_params\", \"tags\", \"workflow_engine_parameters\"):\n                        content = value.read()\n                        body[key] = json.loads(content.decode(\"utf-8\"))\n                    else:\n                        body[key] = value.read().decode()\n            except Exception as e:\n                raise MalformedRequestException(f\"Error reading parameter '{key}': {e}\")\n\n        for key, ls in connexion.request.form.lists():\n            try:\n                for value in ls:\n                    if not value:\n                        continue\n                    if key in (\"workflow_params\", \"tags\", \"workflow_engine_parameters\"):\n                        body[key] = json.loads(value)\n                    else:\n                        body[key] = value\n            except Exception as e:\n                raise MalformedRequestException(f\"Error reading parameter '{key}': {e}\")\n\n        if \"workflow_url\" in body:\n            url, ref = urldefrag(body[\"workflow_url\"])\n            if \":\" not in url:\n                if not has_attachments:\n                    raise MalformedRequestException(\"Relative 'workflow_url' but missing 'workflow_attachment'\")\n                body[\"workflow_url\"] = self.secure_path(url)  # keep this relative\n                if ref:\n                    # append \"#ref\" after the url\n                    body[\"workflow_url\"] += \"#\" + self.secure_path(ref)\n            self.log_for_run(run_id, \"Using workflow_url '%s'\" % body.get(\"workflow_url\"))\n        else:\n            raise MalformedRequestException(\"Missing 'workflow_url' in submission\")\n        if \"workflow_params\" not in body:\n            raise MalformedRequestException(\"Missing 'workflow_params' in submission\")\n\n        return temp_dir, body\n", "path": "src/toil/server/wes/abstract_backend.py"}]}