Fix.

Author: tobiasraabe

.pre-commit-config.yaml

@@ -72,12 +72,13 @@ repos:
             --ignore-missing-imports,
         ]
         additional_dependencies: [
+            cloudpickle,
+            optree,
+            pytask==0.4.0rc2,
+            rich,
             types-attrs,
             types-click,
             types-setuptools,
-            pytask==0.4.0rc2,
-            rich,
-            optree,
         ]
         pass_filenames: false
 -   repo: https://github.com/mgedmin/check-manifest

src/pytask_parallel/backends.py

@@ -9,12 +9,16 @@
 from typing import Callable
 
 import cloudpickle
+from _pytask.path import import_path
 
 
 def deserialize_and_run_with_cloudpickle(
-    fn: Callable[..., Any], kwargs: dict[str, Any]
+    fn: bytes, kwargs: bytes, kwargs_import_path: bytes
 ) -> Any:
     """Deserialize and execute a function and keyword arguments."""
+    deserialized_kwargs_import_path = cloudpickle.loads(kwargs_import_path)
+    import_path(**deserialized_kwargs_import_path)
+
     deserialized_fn = cloudpickle.loads(fn)
     deserialized_kwargs = cloudpickle.loads(kwargs)
     return deserialized_fn(**deserialized_kwargs)
@@ -31,6 +35,7 @@ def submit(  # type: ignore[override]
         return super().submit(
             deserialize_and_run_with_cloudpickle,
             fn=cloudpickle.dumps(fn),
+            kwargs_import_path=cloudpickle.dumps(kwargs.pop("kwargs_import_path")),
             kwargs=cloudpickle.dumps(kwargs),
         )
 

src/pytask_parallel/execute.py

@@ -19,11 +19,13 @@
 from pytask import Mark
 from pytask import parse_warning_filter
 from pytask import PTask
+from pytask import PTaskWithPath
 from pytask import remove_internal_traceback_frames_from_exc_info
 from pytask import Session
 from pytask import Task
 from pytask import warning_record_to_str
 from pytask import WarningReport
+from pytask.tree_util import PyTree
 from pytask.tree_util import tree_leaves
 from pytask.tree_util import tree_map
 from pytask.tree_util import tree_structure
@@ -175,7 +177,7 @@ class ProcessesNameSpace:
 
     @staticmethod
     @hookimpl(tryfirst=True)
-    def pytask_execute_task(session: Session, task: Task) -> Future[Any] | None:
+    def pytask_execute_task(session: Session, task: PTask) -> Future[Any] | None:
         """Execute a task.
 
         Take a task, pickle it and send the bytes over to another process.
@@ -184,6 +186,11 @@ def pytask_execute_task(session: Session, task: Task) -> Future[Any] | None:
         if session.config["n_workers"] > 1:
             kwargs = _create_kwargs_for_task(task)
 
+            if sys.platform == "win32" and isinstance(task, PTaskWithPath):
+                kwargs_import_path = {"path": task.path, "root": session.config["root"]}
+            else:
+                kwargs_import_path = None
+
             return session.config["_parallel_executor"].submit(
                 _execute_task,
                 task=task,
@@ -192,6 +199,7 @@ def pytask_execute_task(session: Session, task: Task) -> Future[Any] | None:
                 console_options=console.options,
                 session_filterwarnings=session.config["filterwarnings"],
                 task_filterwarnings=get_marks(task, "filterwarnings"),
+                kwargs_import_path=kwargs_import_path,
             )
         return None
 
@@ -319,7 +327,7 @@ def _mock_processes_for_threads(
     return [], exc_info
 
 
-def _create_kwargs_for_task(task: Task) -> dict[Any, Any]:
+def _create_kwargs_for_task(task: PTask) -> dict[str, PyTree[Any]]:
     """Create kwargs for task function."""
     parameters = inspect.signature(task.function).parameters
 

tests/test_execute.py

@@ -112,46 +112,6 @@ def task_2(produces):
     assert end - start < 10
 
 
-@pytest.mark.integration()
-@pytest.mark.parametrize("parallel_backend", _PARALLEL_BACKENDS_PARAMETRIZATION)
-def test_pytask_execute_task_w_processes(parallel_backend):
-    # Local function which cannot be used with multiprocessing.
-    def myfunc():
-        return 1
-
-    # Verify it cannot be used with multiprocessing because it cannot be pickled.
-    with pytest.raises(AttributeError):
-        pickle.dumps(myfunc)
-
-    task = Task(base_name="task_example", path=Path(), function=myfunc)
-
-    session = Session()
-    session.config = {
-        "n_workers": 2,
-        "parallel_backend": parallel_backend,
-        "show_locals": False,
-        "filterwarnings": [],
-    }
-
-    with PARALLEL_BACKENDS[parallel_backend](
-        max_workers=session.config["n_workers"]
-    ) as executor:
-        session.config["_parallel_executor"] = executor
-
-        backend_name_space = {
-            ParallelBackendChoices.PROCESSES: ProcessesNameSpace,
-            ParallelBackendChoices.THREADS: DefaultBackendNameSpace,
-            ParallelBackendChoices.LOKY: DefaultBackendNameSpace,
-        }[parallel_backend]
-
-        future = backend_name_space.pytask_execute_task(session, task)
-        executor.shutdown()
-
-    warning_reports, exception = future.result()
-    assert warning_reports == []
-    assert exception is None
-
-
 @pytest.mark.end_to_end()
 @pytest.mark.parametrize("parallel_backend", _PARALLEL_BACKENDS_PARAMETRIZATION)
 def test_stop_execution_when_max_failures_is_reached(tmp_path, parallel_backend):
@@ -318,3 +278,21 @@ def task_example() -> Annotated[str, Path("file.txt")]:
     )
     assert result.exit_code == ExitCode.OK
     assert tmp_path.joinpath("file.txt").exists()
+
+
+
+@pytest.mark.end_to_end()
+@pytest.mark.parametrize("parallel_backend", _PARALLEL_BACKENDS_PARAMETRIZATION)
+def test_task_without_path_that_return(runner, tmp_path, parallel_backend):
+    source = """
+    from pathlib import Path
+    from pytask import task
+
+    task_example = task(produces=Path("file.txt"))(lambda *x: "Hello, Darkness, my old friend.")
+    """
+    tmp_path.joinpath("task_example.py").write_text(textwrap.dedent(source))
+    result = runner.invoke(
+        cli, [tmp_path.as_posix(), "--parallel-backend", parallel_backend]
+    )
+    assert result.exit_code == ExitCode.OK
+    assert tmp_path.joinpath("file.txt").exists()