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/file_formats.py

@@ -6,11 +6,11 @@
 """
 
 
-from rdflib import OWL, RDFS, Graph, URIRef
+from rdflib import Graph, OWL, RDFS, URIRef
 from schema_salad.exceptions import ValidationException
 from schema_salad.utils import aslist, json_dumps
 
-from cwl_utils.types import CWLObjectType
+from cwl_utils.types import CWLFileType
 
 
 def formatSubclassOf(
@@ -49,7 +49,7 @@ def formatSubclassOf(
 
 
 def check_format(
-    actual_file: CWLObjectType | list[CWLObjectType],
+    actual_file: CWLFileType | list[CWLFileType],
     input_formats: list[str] | str,
     ontology: Graph | None,
 ) -> None:

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,22 +28,48 @@
 ]
 
 
-CWLOutputAtomType: TypeAlias = (
-    None
-    | bool
-    | str
-    | int
-    | float
-    | MutableSequence["CWLOutputAtomType"]
-    | MutableMapping[str, "CWLOutputAtomType"]
+CWLDirectoryType = TypedDict(
+    "CWLDirectoryType",
+    {
+        "class": Required[Literal["Directory"]],
+        "location": str,
+        "path": str,
+        "basename": str,
+        "listing": MutableSequence["CWLFileType | CWLDirectoryType"],
+    },
+    total=False,
+)
+
+
+CWLFileType = TypedDict(
+    "CWLFileType",
+    {
+        "class": Required[Literal["File"]],
+        "location": str,
+        "path": str,
+        "basename": str,
+        "dirname": str,
+        "nameroot": str,
+        "nameext": str,
+        "checksum": str,
+        "size": int,
+        "secondaryFiles": MutableSequence["CWLFileType | CWLDirectoryType"],
+        "format": str,
+        "contents": str,
+    },
+    total=False,
 )
+
+
 CWLOutputType: TypeAlias = (
     bool
     | str
     | int
     | float
-    | MutableSequence[CWLOutputAtomType]
-    | MutableMapping[str, CWLOutputAtomType]
+    | CWLFileType
+    | CWLDirectoryType
+    | MutableSequence["CWLOutputType | None"]
+    | MutableMapping[str, "CWLOutputType | None"]
 )
 CWLObjectType: TypeAlias = MutableMapping[str, CWLOutputType | None]
 SinkType: TypeAlias = CWLOutputType | CWLObjectType
@@ -49,9 +82,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 DirentType(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[CWLDirectoryType]:
+    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[CWLFileType]:
+    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[CWLFileType | CWLDirectoryType]:
+    return isinstance(value, Mapping) and value.get("class") in ("File", "Directory")

tests/test_format.py

@@ -14,20 +14,21 @@
 
 from cwl_utils.file_formats import check_format
 from cwl_utils.parser import load_document_by_uri
-from cwl_utils.types import CWLObjectType
-
+from cwl_utils.types import CWLFileType
 from .util import get_path
 
 
-def _create_file(format_: str | None = None) -> CWLObjectType:
-    obj: CWLObjectType = {
-        "class": "File",
-        "basename": "example.txt",
-        "size": 23,
-        "contents": "hoopla",
-        "nameroot": "example",
-        "nameext": "txt",
-    }
+def _create_file(format_: str | None = None) -> CWLFileType:
+    obj = CWLFileType(
+        **{
+            "class": "File",
+            "basename": "example.txt",
+            "size": 23,
+            "contents": "hoopla",
+            "nameroot": "example",
+            "nameext": "txt",
+        }
+    )
     if format_:
         obj["format"] = format_
     return obj
@@ -120,7 +121,7 @@ def test_check_format_no_format() -> None:
 def test_check_format_missing_file() -> None:
     """Confirm that a missing file produces no error."""
     check_format(
-        actual_file=[{}],
+        actual_file=[],
         input_formats="http://edamontology.org/format_2330",
         ontology=EDAM,
     )