[Feature] Validate resource_dict using pydantic (#933)

jan-janssen · pyiron-runner · pre-commit-ci[bot] · web-flow · commit 2fa06c83902a · 2026-02-22T10:54:33.000+01:00
* [Fix] Validate resource_dict using pydantic * Format black * [pre-commit.ci] auto fixes from pre-commit.com hooks for more information, see https://pre-commit.ci * type fixes * fix * fix warnings * fix warning * add validation test * fix test * learn from jules * fix import in test * handle tests on windows * test validation error is raised * validate resource_dict * fix validation * [pre-commit.ci] auto fixes from pre-commit.com hooks for more information, see https://pre-commit.ci * Add cache_key to resource_dict * reduce available keys in resource_dict * [pre-commit.ci] auto fixes from pre-commit.com hooks for more information, see https://pre-commit.ci * always validate resource dict variable * include slurm_cmd_args in validation * fix restart limit check * update documentation * Add remark about pydantic validation * update documentation * [pre-commit.ci] auto fixes from pre-commit.com hooks for more information, see https://pre-commit.ci --------- Co-authored-by: pyiron-runner <pyiron@mpie.de> Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com>
diff --git a/.ci_support/environment-docs.yml b/.ci_support/environment-docs.yml
@@ -15,4 +15,5 @@ dependencies:
 - jupyter-book =1.0.0
 - mpi4py =4.1.1
 - python =3.12
+- pydantic =2.12.4
 - pyzmq =27.1.0
diff --git a/.ci_support/environment-integration.yml b/.ci_support/environment-integration.yml
@@ -17,6 +17,7 @@ dependencies:
 - ipython =9.9.0
 - matplotlib =3.10.8
 - networkx =3.6.1
+- pydantic =2.12.4
 - pygraphviz =1.14
 - pysqa =0.3.5
 - pyzmq =27.1.0
diff --git a/.ci_support/environment-mpich.yml b/.ci_support/environment-mpich.yml
@@ -11,6 +11,7 @@ dependencies:
 - ipython =9.9.0
 - mpi4py =4.1.1
 - networkx =3.6.1
+- pydantic =2.12.4
 - pygraphviz =1.14
 - pysqa =0.3.5
 - pyzmq =27.1.0
diff --git a/.ci_support/environment-old.yml b/.ci_support/environment-old.yml
@@ -10,6 +10,7 @@ dependencies:
 - h5py =3.6.0
 - networkx =2.8.8
 - ipython =7.33.0
+- pydantic =2.5.3
 - pygraphviz =1.10
 - hatchling =1.27.0
 - hatch-vcs =0.4.0
diff --git a/.ci_support/environment-openmpi.yml b/.ci_support/environment-openmpi.yml
@@ -11,6 +11,7 @@ dependencies:
 - ipython =9.9.0
 - mpi4py =4.1.1
 - networkx =3.6.1
+- pydantic =2.12.4
 - pygraphviz =1.14
 - pysqa =0.3.5
 - pyzmq =27.1.0
diff --git a/.ci_support/environment-win.yml b/.ci_support/environment-win.yml
@@ -11,5 +11,6 @@ dependencies:
 - ipython =9.9.0
 - mpi4py =4.1.1
 - networkx =3.6.1
+- pydantic =2.12.4
 - pygraphviz =1.14
 - pyzmq =27.1.0
diff --git a/docs/trouble_shooting.md b/docs/trouble_shooting.md
@@ -42,18 +42,26 @@ The resource dictionary parameter `resource_dict` can contain one or more of the
 * `threads_per_core` (int): number of OpenMP threads to be used for each function call
 * `gpus_per_core` (int): number of GPUs per worker - defaults to 0
 * `cwd` (str/None): current working directory where the parallel python task is executed
-* `openmpi_oversubscribe` (bool): adds the `--oversubscribe` command line flag (OpenMPI and SLURM only) - default False
+* `cache_key` (str): Rather than using the internal hashing of executorlib the user can provide an external `cache_key`
+  to identify tasks on the file system. The initial file name is going to be `cache_key` + `_i.h5` and the final file
+  name is going to be `cache_key` + `_o.h5`.
+* `num_nodes` (int): number of compute nodes used for the evaluation of the Python function. 
+* `exclusive` (bool): boolean flag to reserve exclusive access to selected compute nodes - do not allow other tasks to 
+  use the same compute node. 
+* `error_log_file` (str): path to the error log file, primarily used to merge the log of multiple tasks in one file.
+* `run_time_limit` (int): the maximum time the execution of the submitted Python function is allowed to take in seconds.
+* `priority` (int): the queuing system priority assigned to a given Python function to influence the scheduling.
 * `slurm_cmd_args` (list): Additional command line arguments for the srun call (SLURM only)
 
 For the special case of the [HPC Job Executor](https://executorlib.readthedocs.io/en/latest/3-hpc-job.html) 
 the resource dictionary parameter `resource_dict` can also include additional parameters define in the submission script
 of the [Python simple queuing system adatper (pysqa)](https://pysqa.readthedocs.io) these include but are not limited to: 
-* `run_time_max` (int): the maximum time the execution of the submitted Python function is allowed to take in seconds.
 * `memory_max` (int): the maximum amount of memory the Python function is allowed to use in Gigabytes. 
 * `partition` (str): the partition of the queuing system the Python function is submitted to. 
 * `queue` (str): the name of the queue the Python function is submitted to. 
 
-All parameters in the resource dictionary `resource_dict` are optional. 
+All parameters in the resource dictionary `resource_dict` are optional. When `pydantic` is installed as optional 
+dependency the `resource_dict` is validated using `pydantic`.
 
 ## SSH Connection
 While the [Python simple queuing system adatper (pysqa)](https://pysqa.readthedocs.io) provides the option to connect to
diff --git a/pyproject.toml b/pyproject.toml
@@ -55,10 +55,14 @@ cluster = [
     "pysqa==0.3.5",
     "h5py==3.15.1",
 ]
+validation = [
+    "pydantic==2.12.4",
+]
 all = [
     "mpi4py==4.1.1",
     "pysqa==0.3.5",
     "h5py==3.15.1",
+    "pydantic==2.12.4",
     "pygraphviz==1.14",
     "networkx==3.6.1",
     "ipython==9.9.0",
diff --git a/src/executorlib/executor/flux.py b/src/executorlib/executor/flux.py
diff --git a/src/executorlib/executor/single.py b/src/executorlib/executor/single.py
diff --git a/src/executorlib/executor/slurm.py b/src/executorlib/executor/slurm.py
diff --git a/src/executorlib/standalone/validate.py b/src/executorlib/standalone/validate.py
@@ -0,0 +1,39 @@
+import warnings
+from typing import Optional
+
+from pydantic import BaseModel, Extra
+
+
+class ResourceDictValidation(BaseModel):
+    cores: Optional[int] = None
+    threads_per_core: Optional[int] = None
+    gpus_per_core: Optional[int] = None
+    cwd: Optional[str] = None
+    cache_key: Optional[str] = None
+    num_nodes: Optional[int] = None
+    exclusive: Optional[bool] = None
+    error_log_file: Optional[str] = None
+    run_time_limit: Optional[int] = None
+    priority: Optional[int] = None
+    slurm_cmd_args: Optional[list[str]] = None
+
+    class Config:
+        extra = Extra.forbid
+
+
+def validate_resource_dict(resource_dict: dict) -> None:
+    _ = ResourceDictValidation(**resource_dict)
+
+
+def validate_resource_dict_with_optional_keys(resource_dict: dict) -> None:
+    accepted_keys = ResourceDictValidation.model_fields.keys()
+    optional_lst = [key for key in resource_dict if key not in accepted_keys]
+    validate_dict = {
+        key: value for key, value in resource_dict.items() if key in accepted_keys
+    }
+    _ = ResourceDictValidation(**validate_dict)
+    if len(optional_lst) > 0:
+        warnings.warn(
+            f"The following keys are not recognized and cannot be validated: {optional_lst}",
+            stacklevel=2,
+        )
diff --git a/src/executorlib/task_scheduler/base.py b/src/executorlib/task_scheduler/base.py
@@ -14,6 +14,10 @@
 from executorlib.standalone.serialize import cloudpickle_register
 
 
+def validate_resource_dict(resource_dict: dict):
+    pass
+
+
 class TaskSchedulerBase(FutureExecutor):
     """
     Base class for the executor.
@@ -22,7 +26,11 @@ class TaskSchedulerBase(FutureExecutor):
         max_cores (int): defines the number cores which can be used in parallel
     """
 
-    def __init__(self, max_cores: Optional[int] = None):
+    def __init__(
+        self,
+        max_cores: Optional[int] = None,
+        validator: Callable = validate_resource_dict,
+    ):
         """
         Initialize the ExecutorBase class.
         """
@@ -31,6 +39,7 @@ def __init__(self, max_cores: Optional[int] = None):
         self._max_cores = max_cores
         self._future_queue: Optional[queue.Queue] = queue.Queue()
         self._process: Optional[Union[Thread, list[Thread]]] = None
+        self._validator = validator
 
     @property
     def max_workers(self) -> Optional[int]:
@@ -120,6 +129,7 @@ def submit(  # type: ignore
         """
         if resource_dict is None:
             resource_dict = {}
+        self._validator(resource_dict=resource_dict)
         cores = resource_dict.get("cores")
         if (
             cores is not None
diff --git a/src/executorlib/task_scheduler/file/task_scheduler.py b/src/executorlib/task_scheduler/file/task_scheduler.py
@@ -9,7 +9,7 @@
     check_nested_flux_executor,
     check_pmi_mode,
 )
-from executorlib.task_scheduler.base import TaskSchedulerBase
+from executorlib.task_scheduler.base import TaskSchedulerBase, validate_resource_dict
 from executorlib.task_scheduler.file.shared import execute_tasks_h5
 from executorlib.task_scheduler.file.spawner_subprocess import (
     execute_in_subprocess,
@@ -37,6 +37,7 @@ def __init__(
         pmi_mode: Optional[str] = None,
         wait: bool = True,
         refresh_rate: float = 0.01,
+        validator: Callable = validate_resource_dict,
     ):
         """
         Initialize the FileExecutor.
@@ -55,7 +56,7 @@ def __init__(
             wait (bool): Whether to wait for the completion of all tasks before shutting down the executor.
             refresh_rate (float): The rate at which to refresh the result. Defaults to 0.01.
         """
-        super().__init__(max_cores=None)
+        super().__init__(max_cores=None, validator=validator)
         default_resource_dict = {
             "cores": 1,
             "cwd": None,
@@ -106,6 +107,7 @@ def create_file_executor(
     execute_function: Callable = execute_with_pysqa,
     wait: bool = True,
     refresh_rate: float = 0.01,
+    validator: Callable = validate_resource_dict,
 ):
     if block_allocation:
         raise ValueError(
@@ -138,4 +140,5 @@ def create_file_executor(
         pmi_mode=pmi_mode,
         wait=wait,
         refresh_rate=refresh_rate,
+        validator=validator,
     )
diff --git a/src/executorlib/task_scheduler/interactive/blockallocation.py b/src/executorlib/task_scheduler/interactive/blockallocation.py
@@ -16,7 +16,7 @@
 )
 from executorlib.standalone.interactive.spawner import BaseSpawner, MpiExecSpawner
 from executorlib.standalone.queue import cancel_items_in_queue
-from executorlib.task_scheduler.base import TaskSchedulerBase
+from executorlib.task_scheduler.base import TaskSchedulerBase, validate_resource_dict
 from executorlib.task_scheduler.interactive.shared import (
     execute_task_dict,
     reset_task_dict,
@@ -66,11 +66,14 @@ def __init__(
         max_workers: int = 1,
         executor_kwargs: Optional[dict] = None,
         spawner: type[BaseSpawner] = MpiExecSpawner,
+        validator: Callable = validate_resource_dict,
         restart_limit: int = 0,
     ):
         if executor_kwargs is None:
             executor_kwargs = {}
-        super().__init__(max_cores=executor_kwargs.get("max_cores"))
+        super().__init__(
+            max_cores=executor_kwargs.get("max_cores"), validator=validator
+        )
         executor_kwargs["future_queue"] = self._future_queue
         executor_kwargs["spawner"] = spawner
         executor_kwargs["queue_join_on_shutdown"] = False
diff --git a/src/executorlib/task_scheduler/interactive/dependency.py b/src/executorlib/task_scheduler/interactive/dependency.py
@@ -11,7 +11,7 @@
     get_future_objects_from_input,
     update_futures_in_input,
 )
-from executorlib.task_scheduler.base import TaskSchedulerBase
+from executorlib.task_scheduler.base import TaskSchedulerBase, validate_resource_dict
 from executorlib.task_scheduler.interactive.dependency_plot import (
     export_dependency_graph_function,
     generate_nodes_and_edges_for_plotting,
@@ -47,8 +47,9 @@ def __init__(
         plot_dependency_graph: bool = False,
         plot_dependency_graph_filename: Optional[str] = None,
         export_workflow_filename: Optional[str] = None,
+        validator: Callable = validate_resource_dict,
     ) -> None:
-        super().__init__(max_cores=max_cores)
+        super().__init__(max_cores=max_cores, validator=validator)
         self._process_kwargs = {
             "future_queue": self._future_queue,
             "executor_queue": executor._future_queue,
@@ -136,6 +137,7 @@ def submit(  # type: ignore
         """
         if resource_dict is None:
             resource_dict = {}
+        self._validator(resource_dict=resource_dict)
         if not self._generate_dependency_graph:
             f = super().submit(fn, *args, resource_dict=resource_dict, **kwargs)
         else:
diff --git a/src/executorlib/task_scheduler/interactive/onetoone.py b/src/executorlib/task_scheduler/interactive/onetoone.py
@@ -1,15 +1,15 @@
 import queue
 from concurrent.futures import Future
 from threading import Thread
-from typing import Optional
+from typing import Callable, Optional
 
 from executorlib.standalone.command import get_interactive_execute_command
 from executorlib.standalone.interactive.communication import (
     ExecutorlibSocketError,
     interface_bootup,
 )
 from executorlib.standalone.interactive.spawner import BaseSpawner, MpiExecSpawner
-from executorlib.task_scheduler.base import TaskSchedulerBase
+from executorlib.task_scheduler.base import TaskSchedulerBase, validate_resource_dict
 from executorlib.task_scheduler.interactive.shared import execute_task_dict
 
 
@@ -51,10 +51,13 @@ def __init__(
         max_workers: Optional[int] = None,
         executor_kwargs: Optional[dict] = None,
         spawner: type[BaseSpawner] = MpiExecSpawner,
+        validator: Callable = validate_resource_dict,
     ):
         if executor_kwargs is None:
             executor_kwargs = {}
-        super().__init__(max_cores=executor_kwargs.get("max_cores"))
+        super().__init__(
+            max_cores=executor_kwargs.get("max_cores"), validator=validator
+        )
         executor_kwargs.update(
             {
                 "future_queue": self._future_queue,
diff --git a/src/executorlib/task_scheduler/interactive/spawner_pysqa.py b/src/executorlib/task_scheduler/interactive/spawner_pysqa.py
@@ -11,6 +11,7 @@
     set_current_directory_in_environment,
 )
 from executorlib.standalone.scheduler import pysqa_execute_command, terminate_with_pysqa
+from executorlib.task_scheduler.base import validate_resource_dict
 from executorlib.task_scheduler.interactive.blockallocation import (
     BlockAllocationTaskScheduler,
 )
@@ -230,6 +231,7 @@ def create_pysqa_block_allocation_scheduler(
     executor_kwargs: Optional[dict] = None,
     pysqa_config_directory: Optional[str] = None,
     backend: Optional[str] = None,
+    validator: Callable = validate_resource_dict,
 ):
     if executor_kwargs is None:
         executor_kwargs = {}
@@ -256,4 +258,5 @@ def create_pysqa_block_allocation_scheduler(
         max_workers=max_workers,
         executor_kwargs=executor_kwargs,
         spawner=PysqaSpawner,
+        validator=validator,
     )
diff --git a/tests/unit/executor/test_single_noblock.py b/tests/unit/executor/test_single_noblock.py
@@ -157,7 +157,7 @@ def test_single_node_executor(self):
 
     def test_single_node_executor_block_allocation(self):
         with self.assertRaises(ExecutorlibSocketError):
-            with SingleNodeExecutor(max_workers=2, block_allocation=True, resource_dict={"restart_limit": 2}) as exe:
+            with SingleNodeExecutor(max_workers=2, block_allocation=True, restart_limit=2) as exe:
                 f = exe.submit(exit_funct)
                 print(f.result())
 
diff --git a/tests/unit/standalone/test_validate.py b/tests/unit/standalone/test_validate.py
