Fix lint

pyproject.toml

@@ -30,6 +30,9 @@ ignore = [
     "S107", "S602", "S603", "S608"
 ]
 
+[tool.ruff.lint.pylint]
+max-args=12
+
 [tool.ruff.lint.pydocstyle]
 convention = "google"
 
@@ -95,7 +98,7 @@ ignored-argument-names = '_.*'
 # Maximum number of locals for function / method body
 max-locals = 15
 # Maximum number of return / yield for function / method body
-max-returns = 6
+max-returns = 10
 # Maximum number of branch for function / method body
 max-branches = 12
 # Maximum number of statements in function / method body

src/blue_cwl/core/__init__.py

@@ -1,3 +1,5 @@
 """Core module."""
 
 from blue_cwl.core.parser import parse_cwl_file
+
+__all__ = ["parse_cwl_file"]

src/blue_cwl/core/command.py

@@ -24,6 +24,7 @@ def run_command(
         str_command: command string to execute.
         process_constructor: The process constructor to use. Default is the subprocess.Popen.
         masked_vars: Optional var names to mask when the command is printed.
+        redirect_to: Optional file to redirect the process output.
     """
     if masked_vars is None:
         masked_vars = []

src/blue_cwl/core/config.py

@@ -3,14 +3,16 @@
 from typing import Literal
 
 from blue_cwl.core.common import CustomBaseModel
+from blue_cwl.core.exceptions import CWLError
 
 
 class Config(CustomBaseModel):
     """Config base class."""
 
     def update(self, other):
         """Update config with other config."""
-        assert type(other) is type(self), "Incompatible config types."
+        if type(other) is not type(self):
+            raise CWLError("Incompatible config types.")
         return self.from_dict(self.to_dict() | other.to_dict())
 
 

src/blue_cwl/core/cwl.py

@@ -257,7 +257,6 @@ def get_step_source_names(self, name: str) -> list[str]:
         step = self.get_step_by_name(name)
         sources = set()
         for inp in step.inputs.values():
-
             source_outputs = inp.split_source_output()
 
             if source_outputs is None:

src/blue_cwl/core/cwl_types.py

@@ -5,6 +5,7 @@
 from typing import Literal
 
 from blue_cwl.core.common import CustomBaseModel
+from blue_cwl.core.exceptions import CWLError
 
 CWLType = Literal[
     "null",
@@ -33,7 +34,6 @@ class CWLWorkflowType(enum.Enum):
 
 
 class _FileLike(CustomBaseModel):
-
     path: str | None = None
     location: str | None = None
     basename: str | None = None
@@ -43,7 +43,8 @@ def __init__(self, **data):
         path = data.get("path")
         location = data.get("location")
 
-        assert path or location
+        if path is None and location is None:
+            raise CWLError("Either 'path' or 'location' can be None, not both.")
 
         if path and not location:
             data["path"] = str(path)
@@ -77,6 +78,7 @@ def __init__(self, **data):
         resource_id = data.get("id")
         resource_path = data.get("path")
 
-        assert resource_id or resource_path
+        if resource_id is None and resource_path is None:
+            raise CWLError("Either 'resource_id' or 'resource_path' can be None, not both.")
 
         super().__init__(**data)

src/blue_cwl/core/environment.py

@@ -1,7 +1,7 @@
 """Environment related utilities."""
 
 from pathlib import Path
-from typing import Any, Dict
+from typing import Any
 
 SPACK_MODULEPATH = "/gpfs/bbp.cscs.ch/ssd/apps/bsd/modules/_meta"
 MODULES_ENABLE_PATH = "/etc/profile.d/modules.sh"
@@ -13,7 +13,7 @@
 APPTAINER_IMAGEPATH = "/gpfs/bbp.cscs.ch/ssd/containers"
 
 
-def _build_module_cmd(cmd: str, config: Dict[str, Any]) -> str:
+def _build_module_cmd(cmd: str, config: dict[str, Any]) -> str:
     """Wrap the command with modules."""
     modulepath = config.get("modulepath", SPACK_MODULEPATH)
     modules = config["modules"]
@@ -29,7 +29,7 @@ def _build_module_cmd(cmd: str, config: Dict[str, Any]) -> str:
     )
 
 
-def _build_apptainer_cmd(cmd: str, config: Dict[str, Any]) -> str:
+def _build_apptainer_cmd(cmd: str, config: dict[str, Any]) -> str:
     """Wrap the command with apptainer/singularity."""
     modulepath = config.get("modulepath", APPTAINER_MODULEPATH)
     modules = config.get("modules", APPTAINER_MODULES)
@@ -52,13 +52,13 @@ def _build_apptainer_cmd(cmd: str, config: Dict[str, Any]) -> str:
     return cmd
 
 
-def _build_venv_cmd(cmd: str, config: Dict[str, Any]):
+def _build_venv_cmd(cmd: str, config: dict[str, Any]):
     """Wrap the command with an existing virtual environment."""
     path = config["path"]
     return f". {path}/bin/activate && {cmd}"
 
 
-ENV_MAPPING: Dict[str, Any] = {
+ENV_MAPPING: dict[str, Any] = {
     "MODULE": _build_module_cmd,
     "APPTAINER": _build_apptainer_cmd,
     "VENV": _build_venv_cmd,

src/blue_cwl/core/parser.py

@@ -3,7 +3,7 @@
 import logging
 from copy import deepcopy
 from pathlib import Path
-from typing import Any, Dict, List, Union
+from typing import Any
 
 from blue_cwl.core import config, cwl
 from blue_cwl.core.exceptions import CWLError
@@ -70,7 +70,7 @@ def parse_config_file(cwl_file: PathLike) -> cwl.Config:
 def _preprocess_types(document):
     """Type transformations.
 
-    List of transformations:
+    list of transformations:
         1. Type <T> ending with ? should be transformed to [<T>, "null"].
         2. Type <T> ending with [] should be transformed to {"type": "array", "items": <T>}
 
@@ -92,7 +92,6 @@ def _preprocess_types(document):
 
 
 def _preprocess_type(type_string: str | dict) -> str | list | dict:
-
     if isinstance(type_string, dict):
         return type_string
 
@@ -127,7 +126,6 @@ def _build_workflow(data: dict, base_dir: Path) -> cwl.Workflow:
 
 
 def _preprocess_workflow_document(document, base_dir):
-
     document = deepcopy(document)
     document.pop("class")
 
@@ -150,13 +148,11 @@ def _preprocess_workflow_document(document, base_dir):
 
 
 def _preprocess_workflow_steps(steps, base_dir):
-
     steps = deepcopy(steps)
 
     step_list = _dict_to_list_entries(steps)
 
     for step in step_list:
-
         # in -> inputs
         if "in" in step:
             step["inputs"] = _preprocess_step_inputs(step.pop("in"))
@@ -173,7 +169,6 @@ def _preprocess_workflow_steps(steps, base_dir):
 
 
 def _preprocess_step_inputs(data):
-
     processed = {}
     for k, v in data.items():
         if isinstance(v, str):
@@ -227,7 +222,7 @@ def _preprocess_io(data):
     return {k: {"id": k} | v for k, v in data.items()}
 
 
-def _resolve_paths(raw: Dict[str, Any], base_dir: Path) -> Dict[str, Any]:
+def _resolve_paths(raw: dict[str, Any], base_dir: Path) -> dict[str, Any]:
     """Return a copy of raw data, with paths resolves wrt base_dir."""
 
     def recursive_resolve(entry):
@@ -247,15 +242,15 @@ def recursive_resolve(entry):
     return data
 
 
-def _parse_io_parameters(data: Union[List[Dict[str, Any]], Dict[str, Any]]) -> Dict[str, Any]:
+def _parse_io_parameters(data: list[dict[str, Any]] | dict[str, Any]) -> dict[str, Any]:
     """Return inputs or outputs in dictionary format."""
     if isinstance(data, list):
         return {entry["id"]: {k: v for k, v in entry.items() if k != "id"} for entry in data}
 
     return data
 
 
-def _parse_baseCommand(raw: Union[str, List[str]], base_dir: Path) -> List[str]:
+def _parse_baseCommand(raw: str | list[str], base_dir: Path) -> list[str]:
     base_command = raw
 
     if isinstance(base_command, str):

src/blue_cwl/core/process.py

@@ -140,11 +140,9 @@ def build_command_line_tool_process(
 def _concretize_inputs(
     inputs: dict, input_values: dict[str, InputValue]
 ) -> dict[str, InputValueObject]:
-
     concretized_inputs: dict[str, InputValueObject] = {}
 
     for name, inp in inputs.items():
-
         value = input_values.get(name, None)
         if value is None:
             if inp.default is not None:
@@ -167,7 +165,6 @@ def _concretize_tool_outputs(
 ) -> dict[str, OutputValueObject]:
     concretized_outputs: dict[str, OutputValueObject] = {}
     for name, output in outputs.items():
-
         match output.type:
             case "File":
                 out_binding = output.outputBinding
@@ -222,7 +219,6 @@ def _concretize_tool_command(tool, input_values: dict[str, InputValueObject]) ->
 
 def _cmd_elements(type_, binding, value) -> tuple | list[tuple] | None:
     def _obj_to_string(obj):
-
         match obj:
             case File() | Directory() | NexusResource():
                 return str(obj.path)
@@ -238,18 +234,20 @@ def _separate(prefix, value, separate) -> tuple:
         return (f"{prefix}{value}",)
 
     match type_:
-
         case "File" | "Directory":
             return _separate(binding.prefix, _obj_to_string(value), binding.separate)
 
         case "boolean":
             if bool(value):
-                assert binding.prefix is not None
+                if binding.prefix is None:
+                    raise CWLError("Binding prefix for boolean values cannot be None.")
                 return (binding.prefix,)
             return None
 
         case blue_cwl.core.cwl.CommandInputArraySchema():
-            assert isinstance(value, list)
+            if not isinstance(value, list):
+                raise CWLError(f"Value '{value}' is not a list.")
+
             if item_binding := type_.inputBinding:
                 elements: list[tuple] = []
                 for v in value:
@@ -264,7 +262,8 @@ def _separate(prefix, value, separate) -> tuple:
             return _separate(binding.prefix, value, binding.separate)
 
         case "NexusType":
-            assert isinstance(value.id, str), value
+            if not isinstance(value.id, str):
+                raise CWLError(f"NexusType id '{value.id}' is not a string.")
             return _separate(binding.prefix, value.id, binding.separate)
 
         case _:
@@ -319,7 +318,6 @@ def _get_source_value(source, input_objects, sources):
 
     # The valueFrom fields are evaluated after the the source fields.
     for name, inp in step.inputs.items():
-
         if inp.valueFrom:
             step_input_values[name] = resolve_parameter_references(
                 expression=inp.valueFrom,
@@ -342,7 +340,6 @@ def build_workflow_process(
 
     step_processes: dict[str, CommandLineToolProcess] = {}
     for step in workflow.steps:
-
         sources = {name: step_processes[name] for name in workflow.get_step_source_names(step.id)}
 
         step_processes[step.id] = build_workflow_step_process(
@@ -365,13 +362,16 @@ def build_workflow_process(
 
 
 def _input_value_to_object(input_type, input_value):
-
     match input_type:
-
         case dict():
             # e.g. {"type": "array", "items": "string"}
-            assert input_type["type"] == "array", input_type
-            assert isinstance(input_value, list)
+
+            if input_type["type"] != "array":
+                raise CWLError("Input type is not 'array' for array schema.")
+
+            if not isinstance(input_value, list):
+                raise CWLError(f"Input value '{input_value}' for array is not a list")
+
             element_type = input_type["items"]
             value = [_input_value_to_object(element_type, v) for v in input_value]
 

src/blue_cwl/core/resolve.py

@@ -90,7 +90,7 @@ def walk_dict(keys, dictionary):
                 # e.g. File, Directory
                 current = getattr(current, key)
 
-        if isinstance(current, (File, Directory)):
+        if isinstance(current, File | Directory):
             raise ValueError(f"Parameter resolution should return a string. Got {type(current)}")
 
         return current

src/blue_cwl/core/utils.py

@@ -32,7 +32,7 @@ def wrapper(*args, **kwargs):
         ]
 
         # create argument pairs
-        arg_pairs = [(name, v) for (name, _), v in zip(params[: len(args)], args)]
+        arg_pairs = [(name, v) for (name, _), v in zip(params[: len(args)], args, strict=True)]
 
         # use kwargs or defaults for the rest of the parameters
         arg_pairs.extend(
@@ -71,7 +71,7 @@ def cwd(path):
 @log
 def load_yaml(filepath: PathLike) -> dict:
     """Load from YAML file."""
-    with open(filepath, "r", encoding="utf-8") as f:
+    with open(filepath, encoding="utf-8") as f:
         return yaml.safe_load(f)