Add more TypedDict definitions

This commit adds more `TypedDict` definitions for `File`, `Dirent`, and
`Directory` objects. In addition, it adds the `exitCode` entry to the
`CWLRuntimeParameterContext`, which was missing.

Author: GlassOfWhiskey

cwl_utils/expression.py

@@ -6,14 +6,19 @@
 import json
 from collections.abc import Awaitable, MutableMapping
 from enum import Enum
-from typing import Any, Literal, Union, cast
+from typing import Any, 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, CWLParameterContext
+from cwl_utils.types import (
+    CWLObjectType,
+    CWLOutputType,
+    CWLParameterContext,
+    is_cwl_parameter_context_key,
+)
 from cwl_utils.utils import bytes2str_in_dicts
 
 
@@ -123,19 +128,15 @@ def evaluator(
         if first_symbol_end + 1 == len(ex) and first_symbol == "null":
             return None
         try:
-            if first_symbol in ("inputs", "self", "runtime"):
-                symbol = cast(
-                    Literal["inputs"] | Literal["self"] | Literal["runtime"],
-                    first_symbol,
-                )
+            if is_cwl_parameter_context_key(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]),
+                                cast(CWLOutputType, obj[first_symbol]),
                                 **kwargs,
                             ),
                         )
@@ -146,7 +147,7 @@ def evaluator(
                         js_engine.regex_eval(
                             first_symbol,
                             ex[first_symbol_end:-1],
-                            cast(CWLOutputType, obj[symbol]),
+                            cast(CWLOutputType, obj[first_symbol]),
                             **kwargs,
                         ),
                     )

cwl_utils/sandboxjs.py

@@ -22,7 +22,13 @@
 
 from cwl_utils.errors import JavascriptException, WorkflowException
 from cwl_utils.loghandler import _logger
-from cwl_utils.types import CWLOutputType
+from cwl_utils.types import (
+    CWLOutputType,
+    is_directory,
+    is_directory_key,
+    is_file,
+    is_file_key,
+)
 from cwl_utils.utils import singularity_supports_userns
 
 default_timeout = 20
@@ -549,11 +555,35 @@ def regex_eval(
 
             if isinstance(current_value, Mapping):
                 try:
-                    return self.regex_eval(
-                        parsed_string + remaining_string,
-                        remaining_string[m.end(1) :],
-                        cast(CWLOutputType, current_value[cast(str, key)]),
-                    )
+                    if is_directory(current_value) and is_directory_key(key):
+                        return self.regex_eval(
+                            parsed_string + remaining_string,
+                            remaining_string[m.end(1) :],
+                            cast(
+                                CWLOutputType,
+                                current_value[key],
+                            ),
+                        )
+                    elif is_file(current_value) and is_file_key(key):
+                        return self.regex_eval(
+                            parsed_string + remaining_string,
+                            remaining_string[m.end(1) :],
+                            cast(
+                                CWLOutputType,
+                                current_value[key],
+                            ),
+                        )
+                    else:
+                        return self.regex_eval(
+                            parsed_string + remaining_string,
+                            remaining_string[m.end(1) :],
+                            cast(
+                                CWLOutputType,
+                                cast(MutableMapping[str, Any], current_value)[
+                                    cast(str, key)
+                                ],
+                            ),
+                        )
                 except KeyError as exc:
                     raise WorkflowException(
                         f"{parsed_string!r} doesn't have property {key!r}."

cwl_utils/types.py

@@ -1,8 +1,15 @@
 # SPDX-License-Identifier: Apache-2.0
 # 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 TypeAlias, TypedDict
+import sys
+from collections.abc import Mapping, MutableMapping, MutableSequence
+from typing import Any, Literal, TypeAlias, TypeGuard, TypedDict
+
+if sys.version_info >= (3, 11):
+    from typing import Required
+else:
+    from typing_extensions import Required
+
 
 built_in_types: list[str] = [
     "null",
@@ -21,12 +28,46 @@
 ]
 
 
+DirectoryDict = TypedDict(
+    "DirectoryDict",
+    {
+        "class": Required[Literal["Directory"]],
+        "location": str,
+        "path": str,
+        "basename": str,
+        "listing": MutableSequence["FileDict | DirectoryDict"],
+    },
+    total=False,
+)
+
+
+FileDict = TypedDict(
+    "FileDict",
+    {
+        "class": Required[Literal["File"]],
+        "location": str,
+        "path": str,
+        "basename": str,
+        "dirname": str,
+        "nameroot": str,
+        "nameext": str,
+        "checksum": str,
+        "size": int,
+        "secondaryFiles": MutableSequence["FileDict | DirectoryDict"],
+        "format": str,
+        "contents": str,
+    },
+    total=False,
+)
+
+
 CWLOutputAtomType: TypeAlias = (
     None
     | bool
     | str
     | int
-    | float
+    | FileDict
+    | DirectoryDict
     | MutableSequence["CWLOutputAtomType"]
     | MutableMapping[str, "CWLOutputAtomType"]
 )
@@ -35,6 +76,8 @@
     | str
     | int
     | float
+    | FileDict
+    | DirectoryDict
     | MutableSequence[CWLOutputAtomType]
     | MutableMapping[str, CWLOutputAtomType]
 )
@@ -49,9 +92,76 @@ class CWLRuntimeParameterContext(TypedDict, total=False):
     ram: float | str
     outdirSize: float | str
     tmpdirSize: float | str
+    exitCode: int
 
 
 class CWLParameterContext(TypedDict, total=False):
     inputs: CWLObjectType
     self: CWLOutputType | None
     runtime: CWLRuntimeParameterContext
+
+
+class DirentDict(TypedDict, total=False):
+    entry: Required[str]
+    entryname: str
+    writable: bool
+
+
+def is_cwl_parameter_context_key(
+    key: Any,
+) -> TypeGuard[Literal["inputs", "self", "runtime"]]:
+    return key in ("inputs", "self", "runtime")
+
+
+def is_directory(value: Any) -> TypeGuard[DirectoryDict]:
+    return isinstance(value, Mapping) and value.get("class") == "Directory"
+
+
+def is_directory_key(
+    key: Any,
+) -> TypeGuard[Literal["class", "location", "path", "basename", "listing"]]:
+    return key in ("class", "location", "path", "basename", "listing")
+
+
+def is_file(value: Any) -> TypeGuard[FileDict]:
+    return isinstance(value, Mapping) and value.get("class") == "File"
+
+
+def is_file_key(
+    key: Any,
+) -> TypeGuard[
+    Literal[
+        "class",
+        "location",
+        "path",
+        "basename",
+        "dirname",
+        "nameroot",
+        "nameext",
+        "checksum",
+        "size",
+        "secondaryFiles",
+        "format",
+        "contents",
+    ]
+]:
+    return key in (
+        "class",
+        "location",
+        "path",
+        "basename",
+        "dirname",
+        "nameroot",
+        "nameext",
+        "checksum",
+        "size",
+        "secondaryFiles",
+        "format",
+        "contents",
+    )
+
+
+def is_file_or_directory(
+    value: Any,
+) -> TypeGuard[FileDict | DirectoryDict]:
+    return isinstance(value, Mapping) and value.get("class") in ("File", "Directory")