added getstate and setsate for sge and slurm workers

Author: tclose

docs/source/reference/api.rst

@@ -4,23 +4,23 @@ API
 Python tasks
 ------------
 
-.. automodule:: pydra.design.python
+.. automodule:: pydra.compose.python
    :members:
    :undoc-members:
    :show-inheritance:
 
 Shell tasks
 -----------
 
-.. automodule:: pydra.design.shell
+.. automodule:: pydra.compose.shell
    :members:
    :undoc-members:
    :show-inheritance:
 
 Workflows
 ---------
 
-.. automodule:: pydra.design.workflow
+.. automodule:: pydra.compose.workflow
    :members:
    :undoc-members:
    :show-inheritance:

pydra/workers/sge.py

@@ -78,6 +78,30 @@ class Worker(base.Worker):
     result_files_by_jobid: dict[str, ty.Any] = attrs.field(factory=dict, init=False)
     job_pkls_rerun: dict[str, ty.Any] = attrs.field(factory=dict, init=False)
 
+    _INTERNAL_DICT_ATTRS = [
+        "error",
+        "tasks_to_run_by_threads_requested",
+        "output_by_jobid",
+        "jobid_by_task_uid",
+        "threads_used",
+        "job_completed_by_jobid",
+        "result_files_by_jobid",
+        "job_pkls_rerun",
+    ]
+
+    def __getstate__(self) -> dict[str, ty.Any]:
+        """Return state for pickling."""
+        state = super().__getstate__()
+        for atr in self._INTERNAL_DICT_ATTRS:
+            del state[atr]
+        return state
+
+    def __setstate__(self, state: dict[str, ty.Any]):
+        """Set state for unpickling."""
+        super().__setstate__(state)
+        for atr in self._INTERNAL_DICT_ATTRS:
+            setattr(self, atr, {})
+
     def _prepare_runscripts(self, job, interpreter="/bin/sh", rerun=False):
         if isinstance(job, Job):
             cache_dir = job.cache_dir

pydra/workers/slurm.py

@@ -31,6 +31,17 @@ class Worker(base.Worker):
     sbatch_args: str = ""
     error: dict[str, ty.Any] = attrs.field(factory=dict)
 
+    def __getstate__(self) -> dict[str, ty.Any]:
+        """Return state for pickling."""
+        state = super().__getstate__()
+        del state["error"]
+        return state
+
+    def __setstate__(self, state: dict[str, ty.Any]):
+        """Set state for unpickling."""
+        state["error"] = {}
+        super().__setstate__(state)
+
     def _prepare_runscripts(self, job, interpreter="/bin/sh", rerun=False):
         if isinstance(job, Job):
             cache_dir = job.cache_dir