Add CWLParameterContext as a TypedDict (#392)

GlassOfWhiskey · mr-c · web-flow · commit b9de37839810 · 2025-12-12T17:38:43.000+01:00
This commit further modernizes the CWL type system. It also adds a
`TypedDict` class to represent the CWL parameter context used for
expression evaluation.


Co-authored-by: Michael R. Crusoe &lt;michael.crusoe@gmail.com&gt;
diff --git a/cwl_utils/cwl_v1_0_expression_refactor.py b/cwl_utils/cwl_v1_0_expression_refactor.py
@@ -17,7 +17,12 @@
 import cwl_utils.parser.cwl_v1_0_utils as utils
 from cwl_utils.errors import JavascriptException, WorkflowException
 from cwl_utils.expression import do_eval, interpolate
-from cwl_utils.types import CWLObjectType, CWLOutputType
+from cwl_utils.types import (
+    CWLObjectType,
+    CWLOutputType,
+    CWLParameterContext,
+    CWLRuntimeParameterContext,
+)
 
 
 def expand_stream_shortcuts(process: cwl.CommandLineTool) -> cwl.CommandLineTool:
@@ -85,14 +90,14 @@ def get_expression(
     if string.strip().startswith("${"):
         return string
     if "$(" in string:
-        runtime: CWLObjectType = {
-            "cores": 0,
-            "ram": 0,
-            "outdir": "/root",
-            "tmpdir": "/tmp",  # nosec
-            "outdirSize": 0,
-            "tmpdirSize": 0,
-        }
+        runtime = CWLRuntimeParameterContext(
+            cores=0,
+            ram=0,
+            outdir="/root",
+            tmpdir="/tmp",  # nosec
+            outdirSize=0,
+            tmpdirSize=0,
+        )
         try:
             do_eval(
                 string,
@@ -114,11 +119,9 @@ def get_expression(
                     str,
                     interpolate(
                         scan=string,
-                        rootvars={
-                            "inputs": inputs,
-                            "context": self,
-                            "runtime": runtime,
-                        },
+                        rootvars=CWLParameterContext(
+                            inputs=inputs, self=self, runtime=runtime
+                        ),
                         fullJS=True,
                         escaping_behavior=2,
                         convert_to_expression=True,
diff --git a/cwl_utils/cwl_v1_1_expression_refactor.py b/cwl_utils/cwl_v1_1_expression_refactor.py
@@ -17,7 +17,12 @@
 import cwl_utils.parser.cwl_v1_1_utils as utils
 from cwl_utils.errors import JavascriptException, WorkflowException
 from cwl_utils.expression import do_eval, interpolate
-from cwl_utils.types import CWLObjectType, CWLOutputType
+from cwl_utils.types import (
+    CWLObjectType,
+    CWLOutputType,
+    CWLParameterContext,
+    CWLRuntimeParameterContext,
+)
 
 
 def expand_stream_shortcuts(process: cwl.CommandLineTool) -> cwl.CommandLineTool:
@@ -85,14 +90,14 @@ def get_expression(
     if string.strip().startswith("${"):
         return string
     if "$(" in string:
-        runtime: CWLObjectType = {
-            "cores": 0,
-            "ram": 0,
-            "outdir": "/root",
-            "tmpdir": "/tmp",  # nosec
-            "outdirSize": 0,
-            "tmpdirSize": 0,
-        }
+        runtime = CWLRuntimeParameterContext(
+            cores=0,
+            ram=0,
+            outdir="/root",
+            tmpdir="/tmp",  # nosec
+            outdirSize=0,
+            tmpdirSize=0,
+        )
         try:
             do_eval(
                 string,
@@ -114,11 +119,9 @@ def get_expression(
                     str,
                     interpolate(
                         scan=string,
-                        rootvars={
-                            "inputs": inputs,
-                            "context": self,
-                            "runtime": runtime,
-                        },
+                        rootvars=CWLParameterContext(
+                            inputs=inputs, self=self, runtime=runtime
+                        ),
                         fullJS=True,
                         escaping_behavior=2,
                         convert_to_expression=True,
diff --git a/cwl_utils/cwl_v1_2_expression_refactor.py b/cwl_utils/cwl_v1_2_expression_refactor.py
@@ -17,7 +17,12 @@
 import cwl_utils.parser.cwl_v1_2_utils as utils
 from cwl_utils.errors import JavascriptException, WorkflowException
 from cwl_utils.expression import do_eval, interpolate
-from cwl_utils.types import CWLObjectType, CWLOutputType
+from cwl_utils.types import (
+    CWLObjectType,
+    CWLOutputType,
+    CWLParameterContext,
+    CWLRuntimeParameterContext,
+)
 
 
 def expand_stream_shortcuts(process: cwl.CommandLineTool) -> cwl.CommandLineTool:
@@ -85,14 +90,14 @@ def get_expression(
     if string.strip().startswith("${"):
         return string
     if "$(" in string:
-        runtime: CWLObjectType = {
-            "cores": 0,
-            "ram": 0,
-            "outdir": "/root",
-            "tmpdir": "/tmp",  # nosec
-            "outdirSize": 0,
-            "tmpdirSize": 0,
-        }
+        runtime = CWLRuntimeParameterContext(
+            cores=0,
+            ram=0,
+            outdir="/root",
+            tmpdir="/tmp",  # nosec
+            outdirSize=0,
+            tmpdirSize=0,
+        )
         try:
             do_eval(
                 string,
@@ -114,11 +119,9 @@ def get_expression(
                     str,
                     interpolate(
                         scan=string,
-                        rootvars={
-                            "inputs": inputs,
-                            "context": self,
-                            "runtime": runtime,
-                        },
+                        rootvars=CWLParameterContext(
+                            inputs=inputs, self=self, runtime=runtime
+                        ),
                         fullJS=True,
                         escaping_behavior=2,
                         convert_to_expression=True,
diff --git a/cwl_utils/expression.py b/cwl_utils/expression.py
@@ -6,14 +6,14 @@
 import json
 from collections.abc import Awaitable, MutableMapping
 from enum import Enum
-from typing import Any, Union, cast
+from typing import Any, Literal, Union, cast
 
 from schema_salad.utils import json_dumps
 
 from cwl_utils.errors import JavascriptException, SubstitutionError, WorkflowException
 from cwl_utils.loghandler import _logger
 from cwl_utils.sandboxjs import JSEngine, default_timeout, get_js_engine, param_re
-from cwl_utils.types import CWLObjectType, CWLOutputType
+from cwl_utils.types import CWLObjectType, CWLOutputType, CWLParameterContext
 from cwl_utils.utils import bytes2str_in_dicts
 
 
@@ -108,7 +108,7 @@ def scanner(scan: str) -> tuple[int, int] | None:
 def evaluator(
     js_engine: JSEngine,
     ex: str,
-    obj: CWLObjectType,
+    obj: CWLParameterContext,
     jslib: str,
     fullJS: bool,
     **kwargs: Any,
@@ -123,31 +123,35 @@ def evaluator(
         if first_symbol_end + 1 == len(ex) and first_symbol == "null":
             return None
         try:
-            if first_symbol not in obj:
-                raise WorkflowException("%s is not defined" % first_symbol)
-
-            if inspect.iscoroutinefunction(js_engine.regex_eval):
-                return asyncio.get_event_loop().run_until_complete(
-                    cast(
-                        Awaitable[CWLOutputType],
+            if first_symbol in ("inputs", "self", "runtime"):
+                symbol = cast(
+                    Literal["inputs"] | Literal["self"] | Literal["runtime"],
+                    first_symbol,
+                )
+                if inspect.iscoroutinefunction(js_engine.regex_eval):
+                    return asyncio.get_event_loop().run_until_complete(
+                        cast(
+                            Awaitable[CWLOutputType],
+                            js_engine.regex_eval(
+                                first_symbol,
+                                ex[first_symbol_end:-1],
+                                cast(CWLOutputType, obj[symbol]),
+                                **kwargs,
+                            ),
+                        )
+                    )
+                else:
+                    return cast(
+                        CWLOutputType,
                         js_engine.regex_eval(
                             first_symbol,
                             ex[first_symbol_end:-1],
-                            cast(CWLOutputType, obj[first_symbol]),
+                            cast(CWLOutputType, obj[symbol]),
                             **kwargs,
                         ),
                     )
-                )
             else:
-                return cast(
-                    CWLOutputType,
-                    js_engine.regex_eval(
-                        first_symbol,
-                        ex[first_symbol_end:-1],
-                        cast(CWLOutputType, obj[first_symbol]),
-                        **kwargs,
-                    ),
-                )
+                raise WorkflowException(f"{first_symbol} is unexpected.")
         except WorkflowException as werr:
             expression_parse_exception = werr
     if fullJS:
@@ -174,7 +178,7 @@ def evaluator(
 
 def interpolate(
     scan: str,
-    rootvars: CWLObjectType,
+    rootvars: CWLParameterContext,
     jslib: str = "",
     fullJS: bool = False,
     strip_whitespace: bool = True,
@@ -258,7 +262,7 @@ def dump(string: str) -> str:
     return "".join(parts)
 
 
-def jshead(engine_config: list[str], rootvars: CWLObjectType) -> str:
+def jshead(engine_config: list[str], rootvars: CWLParameterContext) -> str:
     """Make sure all the byte strings are converted to str in `rootvars` dict."""
     return "\n".join(
         engine_config
@@ -293,7 +297,7 @@ def do_eval(
     runtime["outdir"] = outdir or None
 
     rootvars = cast(
-        CWLObjectType,
+        CWLParameterContext,
         bytes2str_in_dicts({"inputs": jobinput, "self": context, "runtime": runtime}),
     )
 
diff --git a/cwl_utils/parser/__init__.py b/cwl_utils/parser/__init__.py
@@ -10,8 +10,8 @@
 from schema_salad.exceptions import ValidationException
 from schema_salad.utils import yaml_no_ts
 
-from . import cwl_v1_0, cwl_v1_1, cwl_v1_2
 from ..errors import GraphTargetMissingException
+from . import cwl_v1_0, cwl_v1_1, cwl_v1_2
 
 
 class NoType(ABC):
diff --git a/cwl_utils/types.py b/cwl_utils/types.py
@@ -2,9 +2,9 @@
 # From https://github.com/rabix/sbpack/blob/b8404a0859ffcbe1edae6d8f934e51847b003320/sbpack/lib.py
 """Shared Python type definitions for commons JSON like CWL objects."""
 from collections.abc import MutableMapping, MutableSequence
-from typing import Any, Optional, Union
+from typing import TypeAlias, TypedDict
 
-built_in_types = [
+built_in_types: list[str] = [
     "null",
     "boolean",
     "int",
@@ -21,31 +21,37 @@
 ]
 
 
-CWLOutputAtomType = Union[
-    None,
-    bool,
-    str,
-    int,
-    float,
-    MutableSequence[
-        Union[
-            None, bool, str, int, float, MutableSequence[Any], MutableMapping[str, Any]
-        ]
-    ],
-    MutableMapping[
-        str,
-        Union[
-            None, bool, str, int, float, MutableSequence[Any], MutableMapping[str, Any]
-        ],
-    ],
-]
-CWLOutputType = Union[
-    bool,
-    str,
-    int,
-    float,
-    MutableSequence[CWLOutputAtomType],
-    MutableMapping[str, CWLOutputAtomType],
-]
-CWLObjectType = MutableMapping[str, Optional[CWLOutputType]]
-SinkType = Union[CWLOutputType, CWLObjectType]
+CWLOutputAtomType: TypeAlias = (
+    None
+    | bool
+    | str
+    | int
+    | float
+    | MutableSequence["CWLOutputAtomType"]
+    | MutableMapping[str, "CWLOutputAtomType"]
+)
+CWLOutputType: TypeAlias = (
+    bool
+    | str
+    | int
+    | float
+    | MutableSequence[CWLOutputAtomType]
+    | MutableMapping[str, CWLOutputAtomType]
+)
+CWLObjectType: TypeAlias = MutableMapping[str, CWLOutputType | None]
+SinkType: TypeAlias = CWLOutputType | CWLObjectType
+
+
+class CWLRuntimeParameterContext(TypedDict, total=False):
+    outdir: str
+    tmpdir: str
+    cores: float | str
+    ram: float | str
+    outdirSize: float | str
+    tmpdirSize: float | str
+
+
+class CWLParameterContext(TypedDict, total=False):
+    inputs: CWLObjectType
+    self: CWLOutputType | None
+    runtime: CWLRuntimeParameterContext
