From c2c99ac38d2934f604beadd69fc2839972821d8a Mon Sep 17 00:00:00 2001 From: "dependabot[bot]" <49699333+dependabot[bot]@users.noreply.github.com> Date: Sat, 3 Oct 2026 11:12:53 +0000 Subject: [PATCH 1/3] Update ty requirement from <0.0.81,>=0.0.8 to >=0.0.8,<0.0.85 Updates the requirements on [ty](https://github.com/astral-sh/ty) to permit the latest version. - [Release notes](https://github.com/astral-sh/ty/releases) - [Changelog](https://github.com/astral-sh/ty/blob/main/CHANGELOG.md) - [Commits](https://github.com/astral-sh/ty/compare/0.0.8...0.0.84) --- updated-dependencies: - dependency-name: ty dependency-version: 0.0.84 dependency-type: direct:development ... Signed-off-by: dependabot[bot] --- pyproject.toml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pyproject.toml b/pyproject.toml index af84431..3394f02 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -50,7 +50,7 @@ test = [ ] typing = [ "pytask-parallel", - "ty>=0.0.8,<0.0.81", + "ty>=0.0.8,<0.0.85", {include-group = "coiled"}, {include-group = "dask"}, ] From 02118a4a8632c8ca4e621b10f107e5e2609d361e Mon Sep 17 00:00:00 2001 From: Tobias Raabe Date: Sat, 3 Oct 2026 14:35:59 +0200 Subject: [PATCH 2/3] Fix typing for updated ty --- docs/source/conf.py | 2 +- src/pytask_parallel/wrappers.py | 9 ++++++--- 2 files changed, 7 insertions(+), 4 deletions(-) diff --git a/docs/source/conf.py b/docs/source/conf.py index a960390..128ccd0 100644 --- a/docs/source/conf.py +++ b/docs/source/conf.py @@ -34,7 +34,7 @@ # The version, including alpha/beta/rc tags, but not commit hash and datestamps release = version("pytask_parallel") # The short X.Y version. -version = ".".join(release.split(".")[:2]) # ty: ignore[invalid-assignment] +version = ".".join(release.split(".")[:2]) # -- General configuration ------------------------------------------------------------- diff --git a/src/pytask_parallel/wrappers.py b/src/pytask_parallel/wrappers.py index ace745a..9fb3991 100644 --- a/src/pytask_parallel/wrappers.py +++ b/src/pytask_parallel/wrappers.py @@ -290,9 +290,12 @@ def _save_and_carry_over_product( node.save(value) return None - return tree_map_with_path( - _save_and_carry_over_product, - cast("Any", task.produces), + return cast( + "PyTree[CarryOverPath | PythonNode | None]", + tree_map_with_path( + _save_and_carry_over_product, + cast("Any", task.produces), + ), ) From 3ba8e9a7738935e20f56b55add1782db7f7ff63b Mon Sep 17 00:00:00 2001 From: Tobias Raabe Date: Sat, 3 Oct 2026 16:09:36 +0200 Subject: [PATCH 3/3] Handle provisional return products in parallel wrapper --- src/pytask_parallel/wrappers.py | 41 +++++++++++++++++++++------------ tests/test_execute.py | 13 +++++++++++ 2 files changed, 39 insertions(+), 15 deletions(-) diff --git a/src/pytask_parallel/wrappers.py b/src/pytask_parallel/wrappers.py index 9fb3991..3d5446d 100644 --- a/src/pytask_parallel/wrappers.py +++ b/src/pytask_parallel/wrappers.py @@ -18,6 +18,7 @@ from attrs import define from pytask import PNode from pytask import PPathNode +from pytask import PProvisionalNode from pytask import PTask from pytask import PythonNode from pytask import Traceback @@ -255,7 +256,7 @@ def _handle_function_products( raise ValueError(msg) def _save_and_carry_over_product( - path: tuple[Any, ...], node: PNode + path: tuple[Any, ...], node: PNode | PProvisionalNode ) -> CarryOverPath | PythonNode | None: argument = path[0] @@ -275,20 +276,7 @@ def _save_and_carry_over_product( for p in path[1:]: value = value[p] - # If the node is a PythonNode, we need to carry it over to the main process. - if isinstance(node, PythonNode): - node.save(value=value) - return node - - # If the path is local and we are remote, we need to carry over the value to - # the main process as a PythonNode and save it later. - if isinstance(node, PPathNode) and is_local_path(node.path) and remote: - return PythonNode(value=value) - - # If no condition applies, we save the value and do not carry it over. Like a - # remote path to S3. - node.save(value) - return None + return _save_return_product(node, value, remote=remote) return cast( "PyTree[CarryOverPath | PythonNode | None]", @@ -299,6 +287,29 @@ def _save_and_carry_over_product( ) +def _save_return_product( + node: PNode | PProvisionalNode, + value: Any, # noqa: ANN401 + *, + remote: bool, +) -> CarryOverPath | PythonNode | None: + """Save a concrete return product or defer a provisional one to collection.""" + if isinstance(node, PProvisionalNode): + return None + + # Python nodes must be carried back to the main process. + if isinstance(node, PythonNode): + node.save(value=value) + return node + + # Local paths on remote workers are saved by the main process. + if isinstance(node, PPathNode) and is_local_path(node.path) and remote: + return PythonNode(value=value) + + node.save(value) + return None + + def _write_local_files_to_remote( kwargs: dict[str, PyTree[Any]], ) -> dict[str, PyTree[Any]]: diff --git a/tests/test_execute.py b/tests/test_execute.py index f8b919e..4e6a316 100644 --- a/tests/test_execute.py +++ b/tests/test_execute.py @@ -6,12 +6,15 @@ from time import time import pytest +from pytask import DirectoryNode from pytask import ExitCode +from pytask import TaskWithoutPath from pytask import build from pytask import cli from pytask_parallel import ParallelBackend from pytask_parallel.execute import _Sleeper +from pytask_parallel.wrappers import _handle_function_products from tests.conftest import restore_sys_path_and_module_after_test_execution from tests.conftest import skip_if_deadlock @@ -28,6 +31,16 @@ ] +def test_provisional_return_product_is_not_saved() -> None: + task = TaskWithoutPath( + name="provisional", + function=lambda: None, + produces={"return": DirectoryNode()}, + ) + + assert _handle_function_products(task, None) == {"return": None} + + @pytest.mark.parametrize("parallel_backend", _IMPLEMENTED_BACKENDS) def test_parallel_execution(tmp_path, parallel_backend): source = """