Fix types.

Author: tobiasraabe

.github/workflows/main.yml

@@ -18,20 +18,20 @@ on:
 
 jobs:
 
-  build-package:
-    name: Build & verify package
+  run-type-checking:
+
+    name: Run tests for type-checking
     runs-on: ubuntu-latest
 
     steps:
       - uses: actions/checkout@v4
+      - uses: actions/setup-python@v5
         with:
-          fetch-depth: 0
-
-      - uses: hynek/build-and-inspect-python-package@v2
-        id: baipp
-
-    outputs:
-      python-versions: ${{ steps.baipp.outputs.supported_python_classifiers_json_array }}
+          python-version-file: .python-version
+          allow-prereleases: true
+          cache: pip
+      - run: pip install tox-uv
+      - run: tox -e typing
 
   run-tests:
 

.pre-commit-config.yaml

@@ -59,24 +59,6 @@ repos:
   rev: v2.2.6
   hooks:
     -   id: codespell
--   repo: https://github.com/pre-commit/mirrors-mypy
-    rev: 'v1.9.0'
-    hooks:
-    -   id: mypy
-        args: [
-            --no-strict-optional,
-            --ignore-missing-imports,
-        ]
-        additional_dependencies: [
-            attrs,
-            cloudpickle,
-            loky,
-            "git+https://github.com/pytask-dev/pytask.git@main",
-            rich,
-            types-click,
-            types-setuptools,
-        ]
-        pass_filenames: false
 -   repo: meta
     hooks:
     -   id: check-hooks-apply

pyproject.toml

@@ -104,6 +104,7 @@ disallow_untyped_defs = true
 no_implicit_optional = true
 warn_redundant_casts = true
 warn_unused_ignores = true
+ignore_missing_imports = true
 
 [[tool.mypy.overrides]]
 module = "tests.*"

src/pytask_parallel/backends.py

@@ -50,6 +50,7 @@ def _get_dask_executor(n_workers: int) -> Executor:
     from pytask import import_optional_dependency
 
     distributed = import_optional_dependency("distributed")
+    assert distributed  # noqa: S101
     try:
         client = distributed.Client.current()
     except ValueError:

src/pytask_parallel/config.py

@@ -27,7 +27,7 @@ def pytask_parse_config(config: dict[str, Any]) -> None:
         raise ValueError(msg) from None
 
     if config["n_workers"] == "auto":
-        config["n_workers"] = max(os.cpu_count() - 1, 1)
+        config["n_workers"] = max(os.cpu_count() - 1, 1)  # type: ignore[operator]
 
     # If more than one worker is used, and no backend is set, use loky.
     if config["n_workers"] > 1 and config["parallel_backend"] == ParallelBackend.NONE:

src/pytask_parallel/execute.py

@@ -8,6 +8,7 @@
 from typing import Any
 
 import cloudpickle
+from _pytask.node_protocols import PPathNode
 from attrs import define
 from attrs import field
 from pytask import ExecutionReport
@@ -24,6 +25,7 @@
 
 from pytask_parallel.backends import WorkerType
 from pytask_parallel.backends import registry
+from pytask_parallel.typing import CarryOverPath
 from pytask_parallel.typing import is_coiled_function
 from pytask_parallel.utils import create_kwargs_for_task
 from pytask_parallel.utils import get_module
@@ -235,7 +237,7 @@ def pytask_unconfigure() -> None:
 
 
 def _update_carry_over_products(
-    task: PTask, carry_over_products: PyTree[PythonNode | None] | None
+    task: PTask, carry_over_products: PyTree[CarryOverPath | PythonNode | None] | None
 ) -> None:
     """Update products carry over from a another process or remote worker.
 
@@ -245,10 +247,18 @@ def _update_carry_over_products(
 
     """
 
-    def _update_carry_over_node(x: PNode, y: PythonNode | None) -> PNode:
-        if y:
+    def _update_carry_over_node(
+        x: PNode, y: CarryOverPath | PythonNode | None
+    ) -> PNode:
+        if y is None:
+            return x
+        if isinstance(x, PPathNode) and isinstance(y, CarryOverPath):
+            x.path.write_bytes(y.content)
+            return x
+        if isinstance(y, PythonNode):
             x.save(y.load())
-        return x
+            return x
+        raise NotImplementedError
 
     structure_carry_over_products = tree_structure(carry_over_products)
     structure_produces = tree_structure(task.produces)

src/pytask_parallel/nodes.py

@@ -2,33 +2,41 @@
 
 from __future__ import annotations
 
-import os
 import tempfile
 from pathlib import Path
-from typing import TYPE_CHECKING
+from typing import Any
 
 from _pytask.node_protocols import PNode
+from _pytask.node_protocols import PPathNode
 from attrs import define
 
 from pytask_parallel.typing import is_local_path
 
-if TYPE_CHECKING:
-    from pytask import PathNode
-
 
 @define(kw_only=True)
 class RemotePathNode(PNode):
-    """The class for a node which is a path."""
+    """A class to handle path nodes with local paths in remote environments.
+
+    Tasks may use nodes, following :class:`pytask.PPathNode`, with paths pointing to
+    local files. These local files should be automatically available in remote
+    environments so that users do not have to care about running their tasks locally or
+    remotely.
+
+    The :class:`RemotePathNode` allows to send local files over to remote environments
+    and back.
+
+    """
 
     name: str
-    local_path: str
+    node: PPathNode
     signature: str
-    value: str | bytes
+    value: Any
+    is_product: bool
     remote_path: str = ""
     fd: int = -1
 
     @classmethod
-    def from_path_node(cls, node: PathNode, *, is_product: bool) -> RemotePathNode:
+    def from_path_node(cls, node: PPathNode, *, is_product: bool) -> RemotePathNode:
         """Instantiate class from path node."""
         if not is_local_path(node.path):
             msg = "Path is not a local path and does not need to be fixed"
@@ -38,34 +46,31 @@ def from_path_node(cls, node: PathNode, *, is_product: bool) -> RemotePathNode:
 
         return cls(
             name=node.name,
-            local_path=node.path.as_posix(),
+            node=node,
             signature=node.signature,
             value=value,
+            is_product=is_product,
         )
 
     def state(self) -> str | None:
         """Calculate the state of the node."""
         msg = "RemotePathNode does not implement .state()."
         raise NotImplementedError(msg)
 
-    def load(self, is_product: bool = False) -> Path:  # noqa: FBT001, FBT002
+    def load(self, is_product: bool = False) -> Path:  # noqa: ARG002, FBT001, FBT002
         """Load the value."""
         # Create a temporary file to store the value.
-        ext = os.path.splitext(self.local_path)[1]  # noqa: PTH122
-        self.fd, self.remote_path = tempfile.mkstemp(suffix=ext)
+        self.fd, self.remote_path = tempfile.mkstemp(suffix=self.node.path.name)
+        path = Path(self.remote_path)
 
         # If the file is a dependency, store the value in the file.
-        path = Path(self.remote_path)
-        if not is_product:
-            path.write_text(self.value) if isinstance(
-                self.value, str
-            ) else path.write_bytes(self.value)
-        return path
+        if not self.is_product:
+            path.write_bytes(self.value)
+
+        # Patch path in original node and load the node.
+        self.node.path = path
+        return self.node.load(is_product=self.is_product)
 
-    def save(self, value: bytes | str) -> None:
+    def save(self, value: Any) -> None:
         """Save strings or bytes to file."""
-        if isinstance(value, (bytes, str)):
-            self.value = value
-        else:
-            msg = f"'RemotePathNode' can only save 'str' and 'bytes', not {type(value)}"
-            raise TypeError(msg)
+        self.value = value

src/pytask_parallel/typing.py

@@ -3,6 +3,7 @@
 from pathlib import Path
 from pathlib import PosixPath
 from pathlib import WindowsPath
+from typing import NamedTuple
 
 from pytask import PTask
 from upath.implementations.local import FilePath
@@ -18,3 +19,7 @@ def is_coiled_function(task: PTask) -> bool:
 def is_local_path(path: Path) -> bool:
     """Check if a path is local."""
     return isinstance(path, (FilePath, PosixPath, WindowsPath))
+
+
+class CarryOverPath(NamedTuple):
+    content: bytes

src/pytask_parallel/utils.py

@@ -9,12 +9,9 @@
 from typing import Callable
 
 from pytask import NodeLoadError
-from pytask import PathNode
-from pytask import PickleNode
 from pytask import PNode
 from pytask import PPathNode
 from pytask import PProvisionalNode
-from pytask import PythonNode
 from pytask.tree_util import PyTree
 from pytask.tree_util import tree_map_with_path
 
@@ -57,7 +54,7 @@ def parse_future_result(
 
         exc_info = _parse_future_exception(future_exception)
         return WrapperResult(
-            carry_over_products=None,
+            carry_over_products=None,  # type: ignore[arg-type]
             warning_reports=[],
             exc_info=exc_info,
             stdout="",
@@ -86,12 +83,7 @@ def _safe_load(
         and isinstance(node, PPathNode)
         and is_local_path(node.path)
     ):
-        if isinstance(node, PathNode):
-            return RemotePathNode.from_path_node(node, is_product=is_product)
-        if isinstance(node, PickleNode):
-            if is_product:
-                return PythonNode()
-            return node.load()
+        return RemotePathNode.from_path_node(node, is_product=is_product)
 
     try:
         return node.load(is_product=is_product)
@@ -136,7 +128,7 @@ def create_kwargs_for_task(task: PTask, *, remote: bool) -> dict[str, PyTree[Any
 
 def _parse_future_exception(
     exc: BaseException | None,
-) -> tuple[type[BaseException], BaseException, TracebackType] | None:
+) -> tuple[type[BaseException], BaseException, TracebackType | None] | None:
     """Parse a future exception into the format of ``sys.exc_info``."""
     return None if exc is None else (type(exc), exc, exc.__traceback__)
 
@@ -156,5 +148,5 @@ def get_module(func: Callable[..., Any], path: Path | None) -> ModuleType:
         func = func.func
 
     if path:
-        return inspect.getmodule(func, path.as_posix())
-    return inspect.getmodule(func)
+        return inspect.getmodule(func, path.as_posix())  # type: ignore[return-value]
+    return inspect.getmodule(func)  # type: ignore[return-value]