Add new switch task type for running different versions of a task #83

Also:
- make sys and os available in all script task expressions
- refactor how named_args are accessed and added to the env
- refactor how captured task results are stashed and accessed
- clean up documentation of task types.

poethepoet/context.py

@@ -58,26 +58,36 @@ def get_task_env(
 
         # Include env vars from dependencies
         if task_uses is not None:
-            result.update(self.get_dep_values(task_uses))
+            result.update(self._get_dep_values(task_uses))
 
         return result
 
-    def get_dep_values(
+    def _get_dep_values(
         self, used_task_invocations: Mapping[str, Tuple[str, ...]]
     ) -> Dict[str, str]:
         """
         Get env vars from upstream tasks declared via the uses option.
-
-        New lines are replaced with whitespace similar to how unquoted command
-        interpolation works in bash.
         """
         return {
-            var_name: re.sub(
-                r"\s+", " ", self.captured_stdout[invocation].strip("\r\n")
-            )
+            var_name: self.get_task_output(invocation)
             for var_name, invocation in used_task_invocations.items()
         }
 
+    def save_task_output(self, invocation: Tuple[str, ...], captured_stdout: bytes):
+        """
+        Store the stdout data from a task so that it can be reused by other tasks
+        """
+        self.captured_stdout[invocation] = captured_stdout.decode()
+
+    def get_task_output(self, invocation: Tuple[str, ...]):
+        """
+        Get the stored stdout data from a task so that it can be reused by other tasks
+
+        New lines are replaced with whitespace similar to how unquoted command
+        interpolation works in bash.
+        """
+        return re.sub(r"\s+", " ", self.captured_stdout[invocation].strip("\r\n"))
+
     def get_executor(
         self,
         invocation: Tuple[str, ...],

poethepoet/env/manager.py

@@ -1,5 +1,5 @@
 from pathlib import Path
-from typing import TYPE_CHECKING, Dict, Mapping, Optional, Union
+from typing import TYPE_CHECKING, Any, Dict, Mapping, Optional, Union
 
 from .cache import EnvFileCache
 from .template import apply_envvars_to_template
@@ -91,8 +91,17 @@ def for_task(
 
         return result
 
-    def update(self, env_vars: Mapping[str, str]):
-        self._vars.update(env_vars)
+    def update(self, env_vars: Mapping[str, Any]):
+        # ensure all values are strings
+        str_vars: Dict[str, str] = {}
+        for key, value in env_vars.items():
+            if isinstance(value, list):
+                str_vars[key] = " ".join(str(item) for item in value)
+            elif value is not None:
+                str_vars[key] = str(value)
+
+        self._vars.update(str_vars)
+
         return self
 
     def to_dict(self):

poethepoet/executor/base.py

@@ -217,7 +217,7 @@ def handle_signal(signum, _frame):
         (captured_stdout, _) = proc.communicate(input)
 
         if self.capture_stdout == True:
-            self.context.captured_stdout[self.invocation] = captured_stdout.decode()
+            self.context.save_task_output(self.invocation, captured_stdout)
 
         # restore signal handler
         signal.signal(signal.SIGINT, old_signal_handler)

poethepoet/helpers/python.py

@@ -31,6 +31,7 @@
     "next",
     "oct",
     "ord",
+    "os",
     "pow",
     "repr",
     "round",
@@ -58,6 +59,7 @@
     "set",
     "slice",
     "str",
+    "sys",
     "tuple",
     "type",
     "zip",
@@ -95,14 +97,11 @@ def resolve_function_call(
         ),
     )
     for node in name_nodes:
-        if node.id in _BUILTINS_WHITELIST:
-            # builtin values have precedence over unqualified args
-            continue
         if node.id in arguments:
             substitutions.append(
                 (_get_name_node_abs_range(source, node), args_prefix + node.id)
             )
-        else:
+        elif node.id not in _BUILTINS_WHITELIST:
             raise ScriptParseError(
                 "Invalid variable reference in script: "
                 + _get_name_source_segment(source, node)

poethepoet/task/__init__.py

@@ -4,3 +4,4 @@
 from .script import ScriptTask
 from .sequence import SequenceTask
 from .shell import ShellTask
+from .switch import SwitchTask

poethepoet/task/base.py

@@ -5,6 +5,7 @@
 from typing import (
     TYPE_CHECKING,
     Any,
+    Collection,
     Dict,
     Iterator,
     List,
@@ -209,26 +210,14 @@ def _parse_named_args(
             return PoeTaskArgs(args_def, self.name, env).parse(extra_args)
         return None
 
-    # @property
-    # def has_named_args(self):
-    #     return bool(self.named_args)
-
     def get_named_arg_values(self, env: EnvVarsManager) -> Dict[str, str]:
-        result: Dict[str, str] = {}
-
         if self.named_args is None:
             self.named_args = self._parse_named_args(self.invocation[1:], env)
 
         if not self.named_args:
             return {}
 
-        for key, value in self.named_args.items():
-            if isinstance(value, list):
-                result[key] = " ".join(str(item) for item in value)
-            elif value is not None:
-                result[key] = str(value)
-
-        return result
+        return self.named_args
 
     def run(
         self,
@@ -320,6 +309,7 @@ def validate_def(
         config: "PoeConfig",
         *,
         anonymous: bool = False,
+        extra_options: Collection[str] = tuple(),
     ) -> Optional[str]:
         """
         Check the given task name and definition for validity and return a message
@@ -350,31 +340,32 @@ def validate_def(
             task_type = cls.__task_types[task_type_key]
             if not isinstance(task_content, task_type.__content_type__):
                 return (
-                    f"Invalid task: {task_name!r}. {task_type} value must be a "
-                    f"{task_type.__content_type__}"
+                    f"Invalid task: {task_name!r}. Content for {task_type.__name__} "
+                    f"must be a {task_type.__content_type__.__name__}"
                 )
-            else:
-                for key in set(task_def) - {task_type_key}:
-                    expected_type = cls.__base_options.get(
-                        key, task_type.__options__.get(key)
-                    )
-                    if expected_type is None:
+
+            for key in set(task_def) - {task_type_key}:
+                expected_type = cls.__base_options.get(
+                    key, task_type.__options__.get(key)
+                )
+                if expected_type is None:
+                    if key not in extra_options:
                         return (
                             f"Invalid task: {task_name!r}. Unrecognised option "
                             f"{key!r} for task of type: {task_type_key}."
                         )
-                    elif not isinstance(task_def[key], expected_type):
-                        return (
-                            f"Invalid task: {task_name!r}. Option {key!r} should "
-                            f"have a value of type {expected_type!r}"
-                        )
-                else:
-                    if hasattr(task_type, "_validate_task_def"):
-                        task_type_issue = task_type._validate_task_def(
-                            task_name, task_def, config
-                        )
-                        if task_type_issue:
-                            return task_type_issue
+                elif not isinstance(task_def[key], expected_type):
+                    return (
+                        f"Invalid task: {task_name!r}. Option {key!r} should "
+                        f"have a value of type {expected_type!r}"
+                    )
+            else:
+                if hasattr(task_type, "_validate_task_def"):
+                    task_type_issue = task_type._validate_task_def(
+                        task_name, task_def, config
+                    )
+                    if task_type_issue:
+                        return task_type_issue
 
             if "args" in task_def:
                 return PoeTaskArgs.validate_def(task_name, task_def["args"])

poethepoet/task/script.py

@@ -32,20 +32,23 @@ def _handle_run(
         extra_args: Sequence[str],
         env: EnvVarsManager,
     ) -> int:
-        # TODO: check whether the project really does use src layout, and don't do
-        #       sys.path.append('src') if it doesn't
-        target_module, function_call = self.parse_script_content(self.named_args)
+        named_arg_values = self.get_named_arg_values(env)
+        env.update(named_arg_values)
+        target_module, function_call = self.parse_script_content(named_arg_values)
         argv = [
             self.name,
             *(env.fill_template(token) for token in extra_args),
         ]
 
+        # TODO: check whether the project really does use src layout, and don't do
+        #       sys.path.append('src') if it doesn't
+
         script = [
-            "import sys; ",
+            "import os,sys; ",
             "from os import environ; ",
             "from importlib import import_module; ",
             f"sys.argv = {argv!r}; sys.path.append('src');",
-            f"{self.format_args_class(self.named_args)}",
+            f"{self.format_args_class(named_arg_values)}",
             f"result = import_module('{target_module}').{function_call};",
         ]