refactor.

Author: tobiasraabe

src/_pytask/collect.py

@@ -597,7 +597,9 @@ def pytask_collect_log(
     """Log collection."""
     session.collection_end = time.time()
 
-    console.print(f"Collected {len(tasks)} task{'' if len(tasks) == 1 else 's'}.")
+    console.print(
+        f"Collected {len(tasks)} task{'' if len(tasks) == 1 else 's'}.", highlight=False
+    )
 
     failed_reports = [r for r in reports if r.outcome == CollectionOutcome.FAIL]
     if failed_reports:

src/_pytask/execute.py

@@ -105,11 +105,7 @@ def pytask_execute_task_protocol(session: Session, task: PTask) -> ExecutionRepo
     """Follow the protocol to execute each task."""
     # Initialize explanation for this task if in explain mode
     if session.config.get("explain", False):
-        task._explanation = TaskExplanation(  # type: ignore[attr-defined]
-            task_name=task.name,
-            would_execute=False,
-            reasons=[],
-        )
+        task.attributes["explanation"] = TaskExplanation(reasons=[])
 
     session.hook.pytask_execute_task_log_start(session=session, task=task)
     try:
@@ -212,9 +208,8 @@ def pytask_execute_task_setup(session: Session, task: PTask) -> None:  # noqa: C
                     needs_to_be_executed = True
 
     # Update explanation on task if in explain mode
-    if session.config["explain"] and hasattr(task, "_explanation"):
-        task._explanation.would_execute = needs_to_be_executed
-        task._explanation.reasons = change_reasons
+    if session.config["explain"] and "explanation" in task.attributes:
+        task.attributes["explanation"].reasons = change_reasons
 
     if not needs_to_be_executed:
         collect_provisional_products(session, task)
@@ -308,7 +303,6 @@ def pytask_execute_task_process_report(
 
     """
     task = report.task
-    explain_mode = session.config.get("explain", False)
 
     if report.outcome == TaskOutcome.SUCCESS:
         update_states_in_database(session, task.signature)
@@ -342,10 +336,6 @@ def pytask_execute_task_process_report(
         if report.exc_info and isinstance(report.exc_info[1], Exit):  # pragma: no cover
             session.should_stop = True
 
-    # Update explanation with outcome if in explain mode
-    if explain_mode and hasattr(task, "_explanation"):
-        task._explanation.outcome = report.outcome
-
     return True
 
 

src/_pytask/explain.py

@@ -56,29 +56,26 @@ def format(self, verbose: int = 1) -> str:  # noqa: PLR0911
 class TaskExplanation:
     """Represents the explanation for why a task needs to be executed."""
 
-    task_name: str
-    would_execute: bool
-    outcome: TaskOutcome | None = None
     reasons: list[ChangeReason] = field(factory=list)
 
-    def format(self, verbose: int = 1) -> str:
+    def format(self, task_name: str, outcome: TaskOutcome, verbose: int = 1) -> str:
         """Format the task explanation as a string."""
         lines = []
 
-        if self.outcome == TaskOutcome.SKIP_UNCHANGED:
-            lines.append(f"{self.task_name}")
+        if outcome == TaskOutcome.SKIP_UNCHANGED:
+            lines.append(f"{task_name}")
             lines.append("  ✓ No changes detected")
-        elif self.outcome == TaskOutcome.PERSISTENCE:
-            lines.append(f"{self.task_name}")
+        elif outcome == TaskOutcome.PERSISTENCE:
+            lines.append(f"{task_name}")
             lines.append("  • Persisted (products exist, changes ignored)")
-        elif self.outcome == TaskOutcome.SKIP:
-            lines.append(f"{self.task_name}")
+        elif outcome == TaskOutcome.SKIP:
+            lines.append(f"{task_name}")
             lines.append("  • Skipped by marker")
         elif not self.reasons:
-            lines.append(f"{self.task_name}")
+            lines.append(f"{task_name}")
             lines.append("  ✓ No changes detected")
         else:
-            lines.append(f"{self.task_name}")
+            lines.append(f"{task_name}")
             lines.extend(reason.format(verbose) for reason in self.reasons)
 
         return "\n".join(lines)
@@ -116,55 +113,73 @@ def pytask_execute_log_end(session: Session, reports: list[ExecutionReport]) ->
     console.rule(Text("Explanation", style="bold blue"), style="bold blue")
     console.print()
 
-    # Collect all explanations
-    explanations = [
-        report.task._explanation
-        for report in reports
-        if hasattr(report.task, "_explanation")
+    # Collect all reports with explanations
+    reports_with_explanations = [
+        report for report in reports if "explanation" in report.task.attributes
     ]
 
-    if not explanations:
+    if not reports_with_explanations:
         console.print("No tasks require execution - everything is up to date.")
         return
 
     # Group by outcome
-    would_execute = [e for e in explanations if e.would_execute]
+    would_execute = [
+        r
+        for r in reports_with_explanations
+        if r.outcome == TaskOutcome.WOULD_BE_EXECUTED
+    ]
     skipped = [
-        e
-        for e in explanations
-        if not e.would_execute and e.outcome != TaskOutcome.SKIP_UNCHANGED
+        r
+        for r in reports_with_explanations
+        if r.outcome in (TaskOutcome.SKIP, TaskOutcome.SKIP_PREVIOUS_FAILED)
+    ]
+    unchanged = [
+        r for r in reports_with_explanations if r.outcome == TaskOutcome.SKIP_UNCHANGED
     ]
-    unchanged = [e for e in explanations if e.outcome == TaskOutcome.SKIP_UNCHANGED]
 
     verbose = session.config.get("verbose", 1)
 
     if would_execute:
-        # WOULD_BE_EXECUTED has style "success" in TaskOutcome
-        console.print(
+        console.rule(
             Text(
-                "Tasks that would be executed:",
+                "─── Tasks that would be executed",
                 style=TaskOutcome.WOULD_BE_EXECUTED.style,
             ),
+            align="left",
+            style=TaskOutcome.WOULD_BE_EXECUTED.style,
         )
         console.print()
-        for exp in would_execute:
-            console.print(exp.format(verbose))
+        for report in would_execute:
+            explanation = report.task.attributes["explanation"]
+            console.print(explanation.format(report.task.name, report.outcome, verbose))
             console.print()
 
     if skipped:
-        # SKIP has style "skipped" in TaskOutcome
-        console.print(Text("Skipped tasks:", style=TaskOutcome.SKIP.style))
+        console.rule(
+            Text("─── Skipped tasks", style=TaskOutcome.SKIP.style),
+            align="left",
+            style=TaskOutcome.SKIP.style,
+        )
         console.print()
-        for exp in skipped:
-            console.print(exp.format(verbose))
+        for report in skipped:
+            explanation = report.task.attributes["explanation"]
+            console.print(explanation.format(report.task.name, report.outcome, verbose))
             console.print()
 
     if unchanged and verbose >= 2:  # noqa: PLR2004
-        # SKIP_UNCHANGED has style "success" in TaskOutcome
-        console.print(
-            Text("Tasks with no changes:", style=TaskOutcome.SKIP_UNCHANGED.style)
+        console.rule(
+            Text("─── Tasks with no changes", style=TaskOutcome.SKIP_UNCHANGED.style),
+            align="left",
+            style=TaskOutcome.SKIP_UNCHANGED.style,
         )
         console.print()
-        for exp in unchanged:
-            console.print(exp.format(verbose))
+        for report in unchanged:
+            explanation = report.task.attributes["explanation"]
+            console.print(explanation.format(report.task.name, report.outcome, verbose))
             console.print()
+
+    elif unchanged and verbose == 1:
+        console.print(
+            f"{len(unchanged)} task(s) with no changes (use -vv to show details)",
+            highlight=False,
+        )

src/_pytask/persist.py

@@ -40,8 +40,6 @@ def pytask_execute_task_setup(session: Session, task: PTask) -> None:
     dependencies and before the same hook implementation for skipping tasks.
 
     """
-    explain_mode = session.config.get("explain", False)
-
     if has_mark(task, "persist"):
         all_states = [
             (
@@ -65,8 +63,6 @@ def pytask_execute_task_setup(session: Session, task: PTask) -> None:
             )
             if any_node_changed:
                 collect_provisional_products(session, task)
-                if explain_mode and hasattr(task, "_explanation"):
-                    task._explanation.outcome = TaskOutcome.PERSISTENCE
                 raise Persisted
 
 

src/_pytask/skipping.py

@@ -49,21 +49,15 @@ def pytask_parse_config(config: dict[str, Any]) -> None:
 @hookimpl
 def pytask_execute_task_setup(session: Session, task: PTask) -> None:
     """Take a short-cut for skipped tasks during setup with an exception."""
-    explain_mode = session.config.get("explain", False)
-
     is_unchanged = has_mark(task, "skip_unchanged") and not has_mark(
         task, "would_be_executed"
     )
     if is_unchanged and not session.config["force"]:
         collect_provisional_products(session, task)
-        if explain_mode and hasattr(task, "_explanation"):
-            task._explanation.outcome = TaskOutcome.SKIP_UNCHANGED
         raise SkippedUnchanged
 
     is_skipped = has_mark(task, "skip")
     if is_skipped:
-        if explain_mode and hasattr(task, "_explanation"):
-            task._explanation.outcome = TaskOutcome.SKIP
         raise Skipped
 
     skipif_marks = get_marks(task, "skipif")
@@ -72,8 +66,6 @@ def pytask_execute_task_setup(session: Session, task: PTask) -> None:
         message = "\n".join(arg[1] for arg in marker_args if arg[0])
         should_skip = any(arg[0] for arg in marker_args)
         if should_skip:
-            if explain_mode and hasattr(task, "_explanation"):
-                task._explanation.outcome = TaskOutcome.SKIP
             raise Skipped(message)
 
     ancestor_failed_marks = get_marks(task, "skip_ancestor_failed")
@@ -82,8 +74,6 @@ def pytask_execute_task_setup(session: Session, task: PTask) -> None:
             skip_ancestor_failed(*mark.args, **mark.kwargs)
             for mark in ancestor_failed_marks
         )
-        if explain_mode and hasattr(task, "_explanation"):
-            task._explanation.outcome = TaskOutcome.SKIP_PREVIOUS_FAILED
         raise SkippedAncestorFailed(message)
 
 

tests/test_explain.py

@@ -266,6 +266,7 @@ def task_example(produces=Path("out.txt")):
         or "unchanged" in result.output.lower()
         or "0 " in result.output
     )
+    assert "1 task(s) with no changes (use -vv to show details)" in result.output
 
 
 def test_explain_with_dry_run(runner, tmp_path):