From 9e0dc65826989402370c667f39d6921a392cc064 Mon Sep 17 00:00:00 2001 From: Tobias Raabe Date: Sun, 2 Aug 2026 17:42:29 +0200 Subject: [PATCH 1/7] Validate unique task signatures --- src/_pytask/collect.py | 3 +++ src/_pytask/dag.py | 2 ++ src/_pytask/task_utils.py | 40 ++++++++++++++++++++++++++++++++ tests/test_collect.py | 48 +++++++++++++++++++++++++++++++++++++++ tests/test_dag.py | 11 +++++++++ 5 files changed, 104 insertions(+) diff --git a/src/_pytask/collect.py b/src/_pytask/collect.py index 7200c8d4..9870e97a 100644 --- a/src/_pytask/collect.py +++ b/src/_pytask/collect.py @@ -53,6 +53,7 @@ from _pytask.task_utils import COLLECTED_TASKS from _pytask.task_utils import parse_collected_tasks_with_task_marker from _pytask.task_utils import task as task_decorator +from _pytask.task_utils import validate_unique_task_signatures from _pytask.typing import TaskFunction from _pytask.typing import is_task_function @@ -81,6 +82,7 @@ def pytask_collect(session: Session) -> bool: try: session.hook.pytask_collect_modify_tasks(session=session, tasks=session.tasks) + validate_unique_task_signatures(session.tasks) except Exception: # noqa: BLE001 # pragma: no cover report = CollectionReport.from_exception( outcome=CollectionOutcome.FAIL, exc_info=sys.exc_info() @@ -605,6 +607,7 @@ def pytask_collect_modify_tasks(tasks: list[PTask]) -> None: for task in tasks: if task.name in id_to_short_id and isinstance(task, Task): task.name = id_to_short_id[task.name] + validate_unique_task_signatures(tasks) def _find_shortest_uniquely_identifiable_name_for_tasks( diff --git a/src/_pytask/dag.py b/src/_pytask/dag.py index 58a53aa6..cf90010f 100644 --- a/src/_pytask/dag.py +++ b/src/_pytask/dag.py @@ -28,6 +28,7 @@ from _pytask.nodes import PythonNode from _pytask.reports import DagReport from _pytask.shared import reduce_names_of_multiple_nodes +from _pytask.task_utils import validate_unique_task_signatures from _pytask.tree_util import tree_map if TYPE_CHECKING: @@ -64,6 +65,7 @@ def create_dag_from_session(session: Session) -> DAG: def _create_dag_from_tasks(tasks: list[PTask]) -> DAG: """Create the DAG from tasks, dependencies and products.""" + validate_unique_task_signatures(tasks) def _add_node_data(dag: DAG, node: PNode | PProvisionalNode) -> None: dag.add_node(node.signature, node) diff --git a/src/_pytask/task_utils.py b/src/_pytask/task_utils.py index 34b2d1e1..44646be6 100644 --- a/src/_pytask/task_utils.py +++ b/src/_pytask/task_utils.py @@ -35,6 +35,8 @@ from ty_extensions import Intersection + from _pytask.node_protocols import PTask + TaskDecorated: TypeAlias = "Intersection[T, TaskFunction]" T = TypeVar("T", bound="Callable[..., Any]") @@ -50,6 +52,7 @@ def _is_task_decorator_target(obj: object) -> TypeGuard[Callable[..., Any]]: "parse_collected_tasks_with_task_marker", "parse_keyword_arguments_from_signature_defaults", "task", + "validate_unique_task_signatures", ] @@ -63,6 +66,43 @@ def _is_task_decorator_target(obj: object) -> TypeGuard[Callable[..., Any]]: """ +def validate_unique_task_signatures(tasks: list[PTask]) -> None: + """Raise an error if multiple tasks have the same signature.""" + signature_to_tasks: dict[str, list[PTask]] = defaultdict(list) + for task_ in tasks: + signature_to_tasks[task_.signature].append(task_) + + collisions = { + signature: tasks_ + for signature, tasks_ in signature_to_tasks.items() + if len(tasks_) > 1 + } + if not collisions: + return + + lines = ["Task signatures must be unique, but pytask collected conflicting tasks."] + for signature, conflicting_tasks in sorted(collisions.items()): + lines.extend(("", f"Signature {signature!r} is used by:")) + lines.extend(f"- {_describe_task(task_)}" for task_ in conflicting_tasks) + + raise ValueError("\n".join(lines)) + + +def _describe_task(task_: PTask) -> str: + """Return a task description which distinguishes conflicting definitions.""" + function_name = getattr(task_.function, "__qualname__", repr(task_.function)) + path = getattr(task_, "path", None) + try: + line_number = inspect.getsourcelines(task_.function)[1] + except (OSError, TypeError): + line_number = None + + location = str(path) if path is not None else "" + if line_number is not None: + location = f"{location}:{line_number}" + return f"{task_.name!r} ({function_name}, {location})" + + @overload def task( name: T, diff --git a/tests/test_collect.py b/tests/test_collect.py index 3b6587e7..52debe7a 100644 --- a/tests/test_collect.py +++ b/tests/test_collect.py @@ -12,17 +12,65 @@ from _pytask.collect import _find_shortest_uniquely_identifiable_name_for_tasks from _pytask.collect import pytask_collect_node from _pytask.node_protocols import PPathNode +from _pytask.task_utils import validate_unique_task_signatures from pytask import CollectionOutcome from pytask import ExitCode from pytask import NodeInfo from pytask import PickleNode from pytask import Session from pytask import Task +from pytask import TaskWithoutPath from pytask import build from pytask import cli from tests.conftest import noop +def test_duplicate_programmatic_task_signatures_fail_collection(tmp_path): + calls = [] + + def first(): + calls.append("first") + + def second(): + calls.append("second") + + tasks = [ + TaskWithoutPath(name="duplicate", function=first), + TaskWithoutPath(name="duplicate", function=second), + ] + + session = build(tasks=tasks, paths=tmp_path, force=True) + + assert session.exit_code == ExitCode.COLLECTION_FAILED + assert not session.execution_reports + assert not calls + + +def test_duplicate_task_signature_message_contains_all_collision_groups(): + def first(): + pass + + def second(): + pass + + tasks = [ + TaskWithoutPath(name="duplicate-a", function=first), + TaskWithoutPath(name="duplicate-a", function=second), + TaskWithoutPath(name="duplicate-b", function=first), + TaskWithoutPath(name="duplicate-b", function=second), + ] + + with pytest.raises(ValueError, match="Task signatures must be unique") as exc_info: + validate_unique_task_signatures(tasks) + + message = str(exc_info.value) + assert "Task signatures must be unique" in message + assert "duplicate-a" in message + assert "duplicate-b" in message + assert "first" in message + assert "second" in message + + def _make_local_upath_uri(path: Path, protocol: str) -> str: return f"{protocol}:///{path.as_posix().lstrip('/')}" diff --git a/tests/test_dag.py b/tests/test_dag.py index 8a81a090..cc4412d9 100644 --- a/tests/test_dag.py +++ b/tests/test_dag.py @@ -10,6 +10,7 @@ from pytask import ExitCode from pytask import PathNode from pytask import Task +from pytask import TaskWithoutPath from pytask import build from pytask import cli from tests.conftest import noop @@ -37,6 +38,16 @@ def test_create_dag(): assert signature in dag.nodes +def test_create_dag_rejects_duplicate_task_signatures(): + tasks = [ + TaskWithoutPath(name="duplicate", function=noop), + TaskWithoutPath(name="duplicate", function=lambda: None), + ] + + with pytest.raises(ValueError, match="Task signatures must be unique"): + _create_dag_from_tasks(tasks) + + def test_cycle_in_dag(tmp_path, runner, snapshot_cli): source = """ from pathlib import Path From a505d76521085edccfe2a10d9a74cfda4380c36f Mon Sep 17 00:00:00 2001 From: "pre-commit-ci[bot]" <66853113+pre-commit-ci[bot]@users.noreply.github.com> Date: Sun, 2 Aug 2026 15:46:46 +0000 Subject: [PATCH 2/7] [pre-commit.ci] auto fixes from pre-commit.com hooks for more information, see https://pre-commit.ci --- tests/test_mark.py | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/tests/test_mark.py b/tests/test_mark.py index 1fb7713e..62218508 100644 --- a/tests/test_mark.py +++ b/tests/test_mark.py @@ -191,8 +191,10 @@ def task_func(arg=arg): [ ( "foo or", - ("at column 7: expected not OR left parenthesis OR identifier; got end of " - "input"), + ( + "at column 7: expected not OR left parenthesis OR identifier; got end of " + "input" + ), ), ( "foo or or", From d1fb5c0f0afb1da9aee4ddf8c527adbbc84b618d Mon Sep 17 00:00:00 2001 From: Tobias Raabe Date: Sat, 15 Aug 2026 17:12:36 +0200 Subject: [PATCH 3/7] Improve duplicate task signature locations --- src/_pytask/task_utils.py | 17 ++++++++++++++--- tests/test_collect.py | 3 +-- 2 files changed, 15 insertions(+), 5 deletions(-) diff --git a/src/_pytask/task_utils.py b/src/_pytask/task_utils.py index 35029a91..9b157575 100644 --- a/src/_pytask/task_utils.py +++ b/src/_pytask/task_utils.py @@ -23,6 +23,7 @@ from _pytask.coiled_utils import Function from _pytask.coiled_utils import extract_coiled_function_kwargs +from _pytask.console import format_task_name from _pytask.console import get_file from _pytask.mark import Mark from _pytask.models import CollectionMetadata @@ -101,17 +102,27 @@ def validate_unique_task_signatures(tasks: list[PTask]) -> None: def _describe_task(task_: PTask) -> str: """Return a task description which distinguishes conflicting definitions.""" - function_name = getattr(task_.function, "__qualname__", repr(task_.function)) path = getattr(task_, "path", None) + if path is None: + try: + path = get_file(task_.function) + except (OSError, TypeError): + path = None try: line_number = inspect.getsourcelines(task_.function)[1] except (OSError, TypeError): line_number = None - location = str(path) if path is not None else "" + if path is None: + location = "" + elif hasattr(path, "as_posix"): + location = path.as_posix() + else: + location = str(path) if line_number is not None: location = f"{location}:{line_number}" - return f"{task_.name!r} ({function_name}, {location})" + task_name = format_task_name(task_, editor_url_scheme="no_link").plain + return f"{task_name} ({location})" @overload diff --git a/tests/test_collect.py b/tests/test_collect.py index 52debe7a..2ad848a5 100644 --- a/tests/test_collect.py +++ b/tests/test_collect.py @@ -67,8 +67,7 @@ def second(): assert "Task signatures must be unique" in message assert "duplicate-a" in message assert "duplicate-b" in message - assert "first" in message - assert "second" in message + assert f"{Path(__file__).as_posix()}:" in message def _make_local_upath_uri(path: Path, protocol: str) -> str: From 1a68c98fdfc54a782a5706ab053b579203bcb8df Mon Sep 17 00:00:00 2001 From: Tobias Raabe Date: Sat, 5 Sep 2026 22:46:05 +0200 Subject: [PATCH 4/7] Validate task identities after collection hooks complete --- src/_pytask/collect.py | 6 +-- src/_pytask/dag.py | 2 - src/_pytask/provisional.py | 7 ++-- src/_pytask/task_utils.py | 41 ++++++++++--------- tests/test_collect.py | 83 ++++++++++++++++++++++++++++++++++++++ tests/test_dag.py | 11 ----- tests/test_provisional.py | 52 ++++++++++++++++++++++++ 7 files changed, 164 insertions(+), 38 deletions(-) diff --git a/src/_pytask/collect.py b/src/_pytask/collect.py index 9870e97a..4d957471 100644 --- a/src/_pytask/collect.py +++ b/src/_pytask/collect.py @@ -51,9 +51,9 @@ from _pytask.shared import to_list from _pytask.shared import unwrap_task_function from _pytask.task_utils import COLLECTED_TASKS +from _pytask.task_utils import modify_and_validate_tasks from _pytask.task_utils import parse_collected_tasks_with_task_marker from _pytask.task_utils import task as task_decorator -from _pytask.task_utils import validate_unique_task_signatures from _pytask.typing import TaskFunction from _pytask.typing import is_task_function @@ -81,8 +81,7 @@ def pytask_collect(session: Session) -> bool: ) try: - session.hook.pytask_collect_modify_tasks(session=session, tasks=session.tasks) - validate_unique_task_signatures(session.tasks) + modify_and_validate_tasks(session) except Exception: # noqa: BLE001 # pragma: no cover report = CollectionReport.from_exception( outcome=CollectionOutcome.FAIL, exc_info=sys.exc_info() @@ -607,7 +606,6 @@ def pytask_collect_modify_tasks(tasks: list[PTask]) -> None: for task in tasks: if task.name in id_to_short_id and isinstance(task, Task): task.name = id_to_short_id[task.name] - validate_unique_task_signatures(tasks) def _find_shortest_uniquely_identifiable_name_for_tasks( diff --git a/src/_pytask/dag.py b/src/_pytask/dag.py index cf90010f..58a53aa6 100644 --- a/src/_pytask/dag.py +++ b/src/_pytask/dag.py @@ -28,7 +28,6 @@ from _pytask.nodes import PythonNode from _pytask.reports import DagReport from _pytask.shared import reduce_names_of_multiple_nodes -from _pytask.task_utils import validate_unique_task_signatures from _pytask.tree_util import tree_map if TYPE_CHECKING: @@ -65,7 +64,6 @@ def create_dag_from_session(session: Session) -> DAG: def _create_dag_from_tasks(tasks: list[PTask]) -> DAG: """Create the DAG from tasks, dependencies and products.""" - validate_unique_task_signatures(tasks) def _add_node_data(dag: DAG, node: PNode | PProvisionalNode) -> None: dag.add_node(node.signature, node) diff --git a/src/_pytask/provisional.py b/src/_pytask/provisional.py index 473b8637..6b19cd9f 100644 --- a/src/_pytask/provisional.py +++ b/src/_pytask/provisional.py @@ -19,6 +19,7 @@ from _pytask.provisional_utils import recreate_dag from _pytask.reports import ExecutionReport from _pytask.task_utils import COLLECTED_TASKS +from _pytask.task_utils import modify_and_validate_tasks from _pytask.task_utils import parse_collected_tasks_with_task_marker from _pytask.tree_util import tree_map from _pytask.tree_util import tree_map_with_path @@ -100,9 +101,7 @@ def pytask_execute_task(session: Session, task: PTask) -> None: ) try: - session.hook.pytask_collect_modify_tasks( - session=session, tasks=session.tasks - ) + modify_and_validate_tasks(session) # Append the last collection report after successful modification if report: session.collection_reports.append(report) @@ -111,6 +110,8 @@ def pytask_execute_task(session: Session, task: PTask) -> None: task=task, exc_info=sys.exc_info() ) session.execution_reports.append(exec_report) + session.should_stop = True + return recreate_dag(session, task) diff --git a/src/_pytask/task_utils.py b/src/_pytask/task_utils.py index 9b157575..d1e71bd6 100644 --- a/src/_pytask/task_utils.py +++ b/src/_pytask/task_utils.py @@ -23,7 +23,6 @@ from _pytask.coiled_utils import Function from _pytask.coiled_utils import extract_coiled_function_kwargs -from _pytask.console import format_task_name from _pytask.console import get_file from _pytask.mark import Mark from _pytask.models import CollectionMetadata @@ -39,6 +38,7 @@ from uuid import UUID from _pytask.node_protocols import PTask + from _pytask.session import Session P = ParamSpec("P") R_co = TypeVar("R_co", covariant=True) @@ -61,6 +61,7 @@ def _is_task_decorator_target(obj: object) -> TypeGuard[Callable[..., Any]]: __all__ = [ "COLLECTED_TASKS", + "modify_and_validate_tasks", "parse_collected_tasks_with_task_marker", "parse_keyword_arguments_from_signature_defaults", "task", @@ -78,6 +79,12 @@ def _is_task_decorator_target(obj: object) -> TypeGuard[Callable[..., Any]]: """ +def modify_and_validate_tasks(session: Session) -> None: + """Validate final task identities after all modification hooks have completed.""" + session.hook.pytask_collect_modify_tasks(session=session, tasks=session.tasks) + validate_unique_task_signatures(session.tasks) + + def validate_unique_task_signatures(tasks: list[PTask]) -> None: """Raise an error if multiple tasks have the same signature.""" signature_to_tasks: dict[str, list[PTask]] = defaultdict(list) @@ -97,32 +104,30 @@ def validate_unique_task_signatures(tasks: list[PTask]) -> None: lines.extend(("", f"Signature {signature!r} is used by:")) lines.extend(f"- {_describe_task(task_)}" for task_ in conflicting_tasks) + lines.extend( + ("", "Assign distinct task names or IDs, or remove the duplicate registration.") + ) raise ValueError("\n".join(lines)) def _describe_task(task_: PTask) -> str: """Return a task description which distinguishes conflicting definitions.""" - path = getattr(task_, "path", None) - if path is None: - try: - path = get_file(task_.function) - except (OSError, TypeError): - path = None + function = unwrap_task_function(task_.function) + while isinstance(function, functools.partial): + function = unwrap_task_function(function.func) + callable_name = getattr(function, "__qualname__", type(function).__qualname__) + module = getattr(function, "__module__", None) + if module: + callable_name = f"{module}.{callable_name}" + try: - line_number = inspect.getsourcelines(task_.function)[1] + path = get_file(function) + line_number = inspect.getsourcelines(function)[1] except (OSError, TypeError): - line_number = None - - if path is None: location = "" - elif hasattr(path, "as_posix"): - location = path.as_posix() else: - location = str(path) - if line_number is not None: - location = f"{location}:{line_number}" - task_name = format_task_name(task_, editor_url_scheme="no_link").plain - return f"{task_name} ({location})" + location = f"{path.as_posix()}:{line_number}" if path else "" + return f"{task_.name!r}: {callable_name} ({location})" @overload diff --git a/tests/test_collect.py b/tests/test_collect.py index 2ad848a5..6227e806 100644 --- a/tests/test_collect.py +++ b/tests/test_collect.py @@ -12,16 +12,19 @@ from _pytask.collect import _find_shortest_uniquely_identifiable_name_for_tasks from _pytask.collect import pytask_collect_node from _pytask.node_protocols import PPathNode +from _pytask.pluginmanager import get_plugin_manager from _pytask.task_utils import validate_unique_task_signatures from pytask import CollectionOutcome from pytask import ExitCode from pytask import NodeInfo +from pytask import PathNode from pytask import PickleNode from pytask import Session from pytask import Task from pytask import TaskWithoutPath from pytask import build from pytask import cli +from pytask import hookimpl from tests.conftest import noop @@ -784,3 +787,83 @@ def test_print_warning_if_non_matching_path_is_passed(runner, tmp_path): assert result.exit_code == ExitCode.OK assert "Collected 0 tasks" in result.output assert "Warning: The path" in result.output + + +@pytest.mark.parametrize("introduce_collision", [True, False]) +def test_validate_task_signatures_after_hook_wrapper( + tmp_path, monkeypatch, introduce_collision +): + class Plugin: + @hookimpl(wrapper=True) + def pytask_collect_modify_tasks(self, tasks): + result = yield + tasks[1].name = "first" if introduce_collision else "second" + return result + + pm = get_plugin_manager() + pm.register(Plugin()) + monkeypatch.setattr("_pytask.build.get_plugin_manager", lambda: pm) + calls = [] + tasks = [ + TaskWithoutPath(name="first", function=lambda: calls.append("first")), + TaskWithoutPath( + name="second" if introduce_collision else "first", + function=lambda: calls.append("second"), + ), + ] + + session = build(tasks=tasks, paths=tmp_path, force=True) + + if introduce_collision: + assert session.exit_code == ExitCode.COLLECTION_FAILED + assert not calls + assert any( + report.exc_info + and "Task signatures must be unique" in str(report.exc_info[1]) + for report in session.collection_reports + ) + else: + assert session.exit_code == ExitCode.OK + assert sorted(calls) == ["first", "second"] + + +def test_duplicate_task_diagnostics_use_callable_source(tmp_path): + tasks = [ + Task(base_name="duplicate", path=tmp_path / "task_other.py", function=noop), + Task(base_name="duplicate", path=tmp_path / "task_other.py", function=noop), + ] + + with pytest.raises(ValueError, match="Task signatures must be unique") as exc_info: + validate_unique_task_signatures(tasks) + + message = str(exc_info.value) + assert "tests.conftest.noop" in message + assert f"{Path(__file__).with_name('conftest.py').as_posix()}:" in message + assert f"({tmp_path.as_posix()}/task_other.py:" not in message + assert "remove the duplicate registration" in message + + +def test_duplicate_task_diagnostics_without_source(): + tasks = [TaskWithoutPath(name="duplicate", function=len) for _ in range(2)] + + with pytest.raises(ValueError, match=r"builtins.len \(\)"): + validate_unique_task_signatures(tasks) + + +def test_unique_tasks_can_share_dependency(tmp_path): + dependency = tmp_path / "input.txt" + dependency.write_text("input") + calls = [] + tasks = [ + TaskWithoutPath( + name=name, + function=lambda path: calls.append(path.read_text()), + depends_on={"path": PathNode(path=dependency)}, + ) + for name in ("first", "second") + ] + + session = build(tasks=tasks, paths=tmp_path, force=True) + + assert session.exit_code == ExitCode.OK + assert calls == ["input", "input"] diff --git a/tests/test_dag.py b/tests/test_dag.py index cc4412d9..8a81a090 100644 --- a/tests/test_dag.py +++ b/tests/test_dag.py @@ -10,7 +10,6 @@ from pytask import ExitCode from pytask import PathNode from pytask import Task -from pytask import TaskWithoutPath from pytask import build from pytask import cli from tests.conftest import noop @@ -38,16 +37,6 @@ def test_create_dag(): assert signature in dag.nodes -def test_create_dag_rejects_duplicate_task_signatures(): - tasks = [ - TaskWithoutPath(name="duplicate", function=noop), - TaskWithoutPath(name="duplicate", function=lambda: None), - ] - - with pytest.raises(ValueError, match="Task signatures must be unique"): - _create_dag_from_tasks(tasks) - - def test_cycle_in_dag(tmp_path, runner, snapshot_cli): source = """ from pathlib import Path diff --git a/tests/test_provisional.py b/tests/test_provisional.py index 203e784f..6de98722 100644 --- a/tests/test_provisional.py +++ b/tests/test_provisional.py @@ -2,10 +2,14 @@ import textwrap +import pytest + +from _pytask.pluginmanager import get_plugin_manager from pytask import ExitCode from pytask import TaskOutcome from pytask import build from pytask import cli +from pytask import hookimpl def test_task_that_produces_provisional_path_node(tmp_path): @@ -280,3 +284,51 @@ def task_example( assert result.exit_code == ExitCode.OK assert tmp_path.joinpath("subfolder", "a.txt").exists() assert tmp_path.joinpath("subfolder", "b.txt").exists() + + +@pytest.mark.parametrize("collision_in_wrapper", [False, True]) +def test_generated_task_identity_conflict_stops_before_dag_rebuild( + tmp_path, monkeypatch, collision_in_wrapper +): + source = """ + from pathlib import Path + from pytask import task, mark + + def task_existing(): + Path(__file__).with_name("existing-ran").touch() + + @mark.try_first + @task(is_generator=True) + def task_generator(): + @task(name=GENERATED_NAME) + def generated(): + Path(__file__).with_name("generated-ran").touch() + """ + generated_name = "task_generated" if collision_in_wrapper else "task_existing" + source = source.replace("GENERATED_NAME", repr(generated_name)) + tmp_path.joinpath("task_module.py").write_text(textwrap.dedent(source)) + + class Plugin: + @hookimpl(wrapper=True) + def pytask_collect_modify_tasks(self, tasks): + result = yield + for task in tasks: + if task.base_name == "task_generated": + task.base_name = "task_existing" + return result + + pm = get_plugin_manager() + pm.register(Plugin()) + monkeypatch.setattr("_pytask.build.get_plugin_manager", lambda: pm) + session = build(paths=tmp_path, force=True) + + assert session.exit_code == ExitCode.FAILED + assert session.should_stop + assert not tmp_path.joinpath("existing-ran").exists() + assert not tmp_path.joinpath("generated-ran").exists() + assert any( + report.exc_info and "Task signatures must be unique" in str(report.exc_info[1]) + for report in session.execution_reports + ) + assert len(session.tasks) == 3 + assert len(session.dag.nodes) == 2 From aa098c2aed27a9ff006b990368d8848230e3f5ec Mon Sep 17 00:00:00 2001 From: Tobias Raabe Date: Sat, 5 Sep 2026 22:57:20 +0200 Subject: [PATCH 5/7] Make task identity errors actionable for each task type --- src/_pytask/task_utils.py | 37 +++++++++++++++++++++++++++++++------ tests/test_collect.py | 32 +++++++++++++++++++++++++++----- tests/test_provisional.py | 2 +- 3 files changed, 59 insertions(+), 12 deletions(-) diff --git a/src/_pytask/task_utils.py b/src/_pytask/task_utils.py index d1e71bd6..9211344d 100644 --- a/src/_pytask/task_utils.py +++ b/src/_pytask/task_utils.py @@ -27,6 +27,8 @@ from _pytask.mark import Mark from _pytask.models import CollectionMetadata from _pytask.models import ParsedAfter +from _pytask.nodes import Task +from _pytask.nodes import TaskWithoutPath from _pytask.shared import find_duplicates from _pytask.shared import unwrap_task_function from _pytask.typing import TaskFunction @@ -99,14 +101,37 @@ def validate_unique_task_signatures(tasks: list[PTask]) -> None: if not collisions: return - lines = ["Task signatures must be unique, but pytask collected conflicting tasks."] - for signature, conflicting_tasks in sorted(collisions.items()): - lines.extend(("", f"Signature {signature!r} is used by:")) + lines = ["Conflicting task identities:"] + groups = sorted( + collisions.values(), key=lambda group: sorted(task_.name for task_ in group) + ) + for conflicting_tasks in groups: + names = ", ".join( + repr(name) for name in sorted({t.name for t in conflicting_tasks}) + ) + lines.extend(("", f"Tasks sharing an identity: {names}")) lines.extend(f"- {_describe_task(task_)}" for task_ in conflicting_tasks) + if all(type(task_) is TaskWithoutPath for task_ in conflicting_tasks): + guidance = ( + "These tasks have the same name. Choose distinct 'name' values " + "for TaskWithoutPath tasks." + ) + elif all(type(task_) is Task for task_ in conflicting_tasks): + guidance = ( + "Each Task must have a unique combination of 'path' and 'base_name'. " + "Choose distinct 'base_name' values for tasks in the same file; " + "changing the display name does not change task identity." + ) + else: + guidance = ( + "These tasks share a signature. For custom task types, ensure the " + "'signature' implementation returns a distinct, stable value for " + "each task identity." + ) + lines.extend( + ("", guidance + " Alternatively, remove the duplicate registration.") + ) - lines.extend( - ("", "Assign distinct task names or IDs, or remove the duplicate registration.") - ) raise ValueError("\n".join(lines)) diff --git a/tests/test_collect.py b/tests/test_collect.py index 6227e806..727b9ce0 100644 --- a/tests/test_collect.py +++ b/tests/test_collect.py @@ -63,13 +63,15 @@ def second(): TaskWithoutPath(name="duplicate-b", function=second), ] - with pytest.raises(ValueError, match="Task signatures must be unique") as exc_info: + with pytest.raises(ValueError, match="Conflicting task identities") as exc_info: validate_unique_task_signatures(tasks) message = str(exc_info.value) - assert "Task signatures must be unique" in message + assert "Conflicting task identities" in message assert "duplicate-a" in message assert "duplicate-b" in message + assert all(task.signature not in message for task in tasks) + assert "Choose distinct 'name' values" in message assert f"{Path(__file__).as_posix()}:" in message @@ -818,8 +820,7 @@ def pytask_collect_modify_tasks(self, tasks): assert session.exit_code == ExitCode.COLLECTION_FAILED assert not calls assert any( - report.exc_info - and "Task signatures must be unique" in str(report.exc_info[1]) + report.exc_info and "Conflicting task identities" in str(report.exc_info[1]) for report in session.collection_reports ) else: @@ -833,7 +834,7 @@ def test_duplicate_task_diagnostics_use_callable_source(tmp_path): Task(base_name="duplicate", path=tmp_path / "task_other.py", function=noop), ] - with pytest.raises(ValueError, match="Task signatures must be unique") as exc_info: + with pytest.raises(ValueError, match="Conflicting task identities") as exc_info: validate_unique_task_signatures(tasks) message = str(exc_info.value) @@ -841,6 +842,8 @@ def test_duplicate_task_diagnostics_use_callable_source(tmp_path): assert f"{Path(__file__).with_name('conftest.py').as_posix()}:" in message assert f"({tmp_path.as_posix()}/task_other.py:" not in message assert "remove the duplicate registration" in message + assert "unique combination of 'path' and 'base_name'" in message + assert "changing the display name does not change task identity" in message def test_duplicate_task_diagnostics_without_source(): @@ -867,3 +870,22 @@ def test_unique_tasks_can_share_dependency(tmp_path): assert session.exit_code == ExitCode.OK assert calls == ["input", "input"] + + +def test_custom_task_identity_conflict_guidance(): + class CustomTask(TaskWithoutPath): + @property + def signature(self): + return "shared-custom-signature" + + tasks = [CustomTask(name=name, function=noop) for name in ("first", "second")] + + with pytest.raises(ValueError, match="Conflicting task identities") as exc_info: + validate_unique_task_signatures(tasks) + + message = str(exc_info.value) + assert "Tasks sharing an identity: 'first', 'second'" in message + assert "shared-custom-signature" not in message + assert "'signature' implementation" in message + assert "distinct, stable value" in message + assert "These tasks have the same name" not in message diff --git a/tests/test_provisional.py b/tests/test_provisional.py index 6de98722..2fcd78a6 100644 --- a/tests/test_provisional.py +++ b/tests/test_provisional.py @@ -327,7 +327,7 @@ def pytask_collect_modify_tasks(self, tasks): assert not tmp_path.joinpath("existing-ran").exists() assert not tmp_path.joinpath("generated-ran").exists() assert any( - report.exc_info and "Task signatures must be unique" in str(report.exc_info[1]) + report.exc_info and "Conflicting task identities" in str(report.exc_info[1]) for report in session.execution_reports ) assert len(session.tasks) == 3 From 087661de421f884a2648ee320826c8c1e9070dce Mon Sep 17 00:00:00 2001 From: Tobias Raabe Date: Sun, 6 Sep 2026 18:23:54 +0200 Subject: [PATCH 6/7] Inline task modification and validation --- src/_pytask/collect.py | 5 +++-- src/_pytask/provisional.py | 7 +++++-- src/_pytask/task_utils.py | 8 -------- 3 files changed, 8 insertions(+), 12 deletions(-) diff --git a/src/_pytask/collect.py b/src/_pytask/collect.py index 4d957471..1d4c697d 100644 --- a/src/_pytask/collect.py +++ b/src/_pytask/collect.py @@ -51,9 +51,9 @@ from _pytask.shared import to_list from _pytask.shared import unwrap_task_function from _pytask.task_utils import COLLECTED_TASKS -from _pytask.task_utils import modify_and_validate_tasks from _pytask.task_utils import parse_collected_tasks_with_task_marker from _pytask.task_utils import task as task_decorator +from _pytask.task_utils import validate_unique_task_signatures from _pytask.typing import TaskFunction from _pytask.typing import is_task_function @@ -81,7 +81,8 @@ def pytask_collect(session: Session) -> bool: ) try: - modify_and_validate_tasks(session) + session.hook.pytask_collect_modify_tasks(session=session, tasks=session.tasks) + validate_unique_task_signatures(session.tasks) except Exception: # noqa: BLE001 # pragma: no cover report = CollectionReport.from_exception( outcome=CollectionOutcome.FAIL, exc_info=sys.exc_info() diff --git a/src/_pytask/provisional.py b/src/_pytask/provisional.py index 6b19cd9f..76f724b0 100644 --- a/src/_pytask/provisional.py +++ b/src/_pytask/provisional.py @@ -19,8 +19,8 @@ from _pytask.provisional_utils import recreate_dag from _pytask.reports import ExecutionReport from _pytask.task_utils import COLLECTED_TASKS -from _pytask.task_utils import modify_and_validate_tasks from _pytask.task_utils import parse_collected_tasks_with_task_marker +from _pytask.task_utils import validate_unique_task_signatures from _pytask.tree_util import tree_map from _pytask.tree_util import tree_map_with_path from _pytask.typing import is_task_generator @@ -101,7 +101,10 @@ def pytask_execute_task(session: Session, task: PTask) -> None: ) try: - modify_and_validate_tasks(session) + session.hook.pytask_collect_modify_tasks( + session=session, tasks=session.tasks + ) + validate_unique_task_signatures(session.tasks) # Append the last collection report after successful modification if report: session.collection_reports.append(report) diff --git a/src/_pytask/task_utils.py b/src/_pytask/task_utils.py index 9211344d..1d0a7eff 100644 --- a/src/_pytask/task_utils.py +++ b/src/_pytask/task_utils.py @@ -40,7 +40,6 @@ from uuid import UUID from _pytask.node_protocols import PTask - from _pytask.session import Session P = ParamSpec("P") R_co = TypeVar("R_co", covariant=True) @@ -63,7 +62,6 @@ def _is_task_decorator_target(obj: object) -> TypeGuard[Callable[..., Any]]: __all__ = [ "COLLECTED_TASKS", - "modify_and_validate_tasks", "parse_collected_tasks_with_task_marker", "parse_keyword_arguments_from_signature_defaults", "task", @@ -81,12 +79,6 @@ def _is_task_decorator_target(obj: object) -> TypeGuard[Callable[..., Any]]: """ -def modify_and_validate_tasks(session: Session) -> None: - """Validate final task identities after all modification hooks have completed.""" - session.hook.pytask_collect_modify_tasks(session=session, tasks=session.tasks) - validate_unique_task_signatures(session.tasks) - - def validate_unique_task_signatures(tasks: list[PTask]) -> None: """Raise an error if multiple tasks have the same signature.""" signature_to_tasks: dict[str, list[PTask]] = defaultdict(list) From 860044c3fac3238709a60e0cb2ee94156b6ae22d Mon Sep 17 00:00:00 2001 From: Tobias Raabe Date: Sun, 6 Sep 2026 18:30:42 +0200 Subject: [PATCH 7/7] add to changes --- CHANGELOG.md | 2 ++ 1 file changed, 2 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index eca39bdf..c0613e3d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -10,6 +10,8 @@ releases are available on [PyPI](https://pypi.org/project/pytask) and - [#889](https://github.com/pytask-dev/pytask/pull/889) improves typing for tree operations by wrapping optree's pytree utilities with pytask-specific signatures and requiring optree 0.16.0 or newer. +- [#932](https://github.com/pytask-dev/pytask/pull/932) validates unique task signatures + during collection. ## 0.6.0 - 2026-05-01