Tasks can contain when deactivated commands.

As long as a `dev-cmd` invocation retains at least 1 command to run, the
run proceeds and just executes fewer commands under the selected Python.

Author: jsirois

CHANGES.md

@@ -1,5 +1,12 @@
 # Release Notes
 
+## 0.28.0
+
+As long as there is at least one command left to execute, tasks and any groups they contain now
+support having individual commands deactivated by `when` environment markers. This allows running
+tasks under multiple different Pythons and just running a subset of commands instead of failing when
+not all commands are available to execute under the selected Python.
+
 ## 0.27.0
 
 The command python value can now be parametrized allowing for tox-style ad-hoc Python specification.

dev_cmd/__init__.py

@@ -1,4 +1,4 @@
 # Copyright 2024 John Sirois.
 # Licensed under the Apache License, Version 2.0 (see LICENSE).
 
-__version__ = "0.27.0"
+__version__ = "0.28.0"

dev_cmd/invoke.py

@@ -12,7 +12,7 @@
 from asyncio.tasks import Task as AsyncTask
 from contextlib import asynccontextmanager
 from dataclasses import dataclass, field
-from typing import Any, AsyncIterator, Container, Mapping
+from typing import Any, AsyncIterator, Container, Iterator, Mapping
 
 from dev_cmd import color
 from dev_cmd.color import USE_COLOR
@@ -65,14 +65,15 @@ def create(
                         f"{accepts_extra_args.name!r} already does."
                     )
                 accepts_extra_args = step.base or step
-            elif command := step.accepts_extra_args(skips):
-                if accepts_extra_args and accepts_extra_args not in (command.base, command):
-                    raise InvalidModelError(
-                        f"The task {step.name!r} invokes command {command.name!r} which accepts extra "
-                        f"args, but only one command can accept extra args per invocation and command "
-                        f"{accepts_extra_args.name!r} already does."
-                    )
-                accepts_extra_args = command.base or command
+            elif commands := tuple(step.accepts_extra_args(skips)):
+                for command in commands:
+                    if accepts_extra_args and accepts_extra_args not in (command.base, command):
+                        raise InvalidModelError(
+                            f"The task {step.name!r} invokes command {command.name!r} which accepts extra "
+                            f"args, but only one command can accept extra args per invocation and command "
+                            f"{accepts_extra_args.name!r} already does."
+                        )
+                    accepts_extra_args = command.base or command
 
         return cls(
             steps=tuple(step for step in steps if step.name not in skips),
@@ -95,6 +96,16 @@ def create(
     console: Console
     _in_flight_processes: dict[Process, Command] = field(default_factory=dict, init=False)
 
+    def iter_commands(self) -> Iterator[Command]:
+        for step in self.steps:
+            if step.name in self.skips:
+                continue
+            elif isinstance(step, Command):
+                yield step
+            else:
+                for command in step.iter_commands(self.skips):
+                    yield command
+
     @asynccontextmanager
     async def _guarded_ctrl_c(self) -> AsyncIterator[None]:
         try:

dev_cmd/model.py

@@ -8,7 +8,7 @@
 from dataclasses import dataclass, field
 from enum import Enum
 from pathlib import PurePath
-from typing import Any, Container, Mapping, MutableMapping
+from typing import Any, Container, Iterator, Mapping, MutableMapping
 
 from packaging.markers import Marker
 
@@ -60,14 +60,14 @@ class Command:
 class Group:
     members: tuple[Command | Task | Group, ...]
 
-    def accepts_extra_args(self, skips: Container[str]) -> Command | None:
+    def iter_commands(self, skips: Container[str]) -> Iterator[Command]:
         for member in self.members:
             if isinstance(member, Command):
-                if member.accepts_extra_args and member.name not in skips:
-                    return member
-            elif command := member.accepts_extra_args(skips):
-                return command
-        return None
+                if member.name not in skips:
+                    yield member
+            else:
+                for command in member.iter_commands(skips):
+                    yield command
 
 
 @dataclass(frozen=True)
@@ -78,10 +78,15 @@ class Task:
     description: str | None = None
     when: Marker | None = None
 
-    def accepts_extra_args(self, skips: Container[str] = ()) -> Command | None:
-        if self.name in skips:
-            return None
-        return self.steps.accepts_extra_args(skips)
+    def accepts_extra_args(self, skips: Container[str] = ()) -> Iterator[Command]:
+        for command in self.iter_commands(skips):
+            if command.accepts_extra_args:
+                yield command
+
+    def iter_commands(self, skips: Container[str]) -> Iterator[Command]:
+        if self.name not in skips:
+            for command in self.steps.iter_commands(skips):
+                yield command
 
 
 class ExitStyle(Enum):

dev_cmd/parse.py

@@ -7,6 +7,7 @@
 import itertools
 import os
 from collections import defaultdict
+from dataclasses import dataclass
 from pathlib import Path
 from typing import Any, Container, Dict, Iterable, Iterator, List, Mapping, Set, cast
 
@@ -79,12 +80,17 @@ def _parse_when(data: dict[str, Any], table_path: str) -> Marker | None:
         )
 
 
+@dataclass(frozen=True)
+class DeactivatedCommand:
+    name: str
+
+
 def _parse_commands(
     commands: dict[str, Any] | None,
     required_steps: dict[str, list[tuple[Factor, ...]]],
     project_dir: Path,
     marker_environment: dict[str, str] | None,
-) -> Iterator[Command]:
+) -> Iterator[Command | DeactivatedCommand]:
     if not commands:
         raise InvalidModelError(
             "There must be at least one entry in the [tool.dev-cmd.commands] table to run "
@@ -280,8 +286,10 @@ def substitute(text: str) -> str:
                     python=substituted_python,
                 )
 
-            if not when or when.evaluate(marker_environment):
-                final_name = f"{name}{factors_suffix}"
+            final_name = f"{name}{factors_suffix}"
+            if when and not when.evaluate(marker_environment):
+                yield DeactivatedCommand(final_name)
+            else:
                 previous_original_name = seen_commands.get(final_name)
                 if previous_original_name and previous_original_name != original_name:
                     raise InvalidModelError(
@@ -312,14 +320,17 @@ def _parse_group(
     group: list[Any],
     all_task_names: Container[str],
     tasks_defined_so_far: Mapping[str, Task],
-    commands: Mapping[str, Command],
+    commands: Mapping[str, Command | DeactivatedCommand],
 ) -> Group:
     members: list[Command | Task | Group] = []
     for index, member in enumerate(group):
         if isinstance(member, str):
             for item in expand(member):
+                command = commands.get(item)
+                if isinstance(command, DeactivatedCommand):
+                    continue
                 try:
-                    members.append(commands.get(item) or tasks_defined_so_far[item])
+                    members.append(command or tasks_defined_so_far[item])
                 except KeyError:
                     if item in all_task_names:
                         raise InvalidModelError(
@@ -362,7 +373,7 @@ def _parse_group(
 
 def _parse_tasks(
     tasks: dict[str, Any] | None,
-    commands: Mapping[str, Command],
+    commands: Mapping[str, Command | DeactivatedCommand],
     marker_environment: dict[str, str] | None,
 ) -> Iterator[Task]:
     if not tasks:
@@ -454,11 +465,13 @@ def _parse_tasks(
 
 
 def _parse_default(
-    default: Any, commands: Mapping[str, Command], tasks: Mapping[str, Task]
+    default: Any, commands: Mapping[str, Command | DeactivatedCommand], tasks: Mapping[str, Task]
 ) -> Command | Task | None:
     if default is None:
         if len(commands) == 1:
-            return next(iter(commands.values()))
+            only_command = next(iter(commands.values()))
+            if isinstance(only_command, Command):
+                return only_command
         return None
 
     if not isinstance(default, str):
@@ -467,8 +480,11 @@ def _parse_default(
             f"{type(default)}."
         )
 
+    if selected_task := tasks.get(default):
+        return selected_task
     try:
-        return tasks.get(default) or commands[default]
+        selected_command = commands[default]
+        return selected_command if isinstance(selected_command, Command) else None
     except KeyError:
         raise InvalidModelError(
             os.linesep.join(
@@ -875,7 +891,7 @@ def pop_dict(key: str, *, path: str) -> dict[str, Any] | None:
             )
             break
 
-    commands = {
+    commands: dict[str, Command | DeactivatedCommand] = {
         cmd.name: cmd
         for cmd in _parse_commands(
             commands_data,
@@ -904,7 +920,7 @@ def pop_dict(key: str, *, path: str) -> dict[str, Any] | None:
         )
 
     return Configuration(
-        commands=tuple(commands.values()),
+        commands=tuple(cmd for cmd in commands.values() if isinstance(cmd, Command)),
         tasks=tuple(tasks.values()),
         default=default,
         exit_style=exit_style,

dev_cmd/run.py

@@ -174,6 +174,11 @@ def _run(
             f"arguments: {shlex.join(extra_args)}"
         )
 
+    if not tuple(invocation.iter_commands()):
+        raise InvalidArgumentError(
+            "All steps in this invocation of `dev-cmd` were deactivated by `when` markers leaving "
+            "nothing to run."
+        )
     invocation = dataclasses.replace(
         invocation, venvs=_ensure_venvs(invocation.steps, config.pythons)
     )
@@ -435,8 +440,10 @@ def _list(
             rendered_task_name = color.color(task.name, fg="magenta", style="bold")
             if config.default == task:
                 rendered_task_name = f"* {rendered_task_name}"
-            if extra_args_cmd := task.accepts_extra_args():
-                extra_args_help = color.magenta(f" (-- extra {extra_args_cmd.args[0]} args ...)")
+            if extra_args_cmds := tuple(task.accepts_extra_args()):
+                extra_args_help = color.magenta(
+                    f" (-- extra {extra_args_cmds[0].args[0]} args ...)"
+                )
                 rendered_task_name = f"{rendered_task_name}{extra_args_help}"
             if task.description:
                 console.print(f"{rendered_task_name}: ")