diff --git a/README.rst b/README.rst index 68e2689fc..e06c8861e 100644 --- a/README.rst +++ b/README.rst @@ -271,6 +271,68 @@ You can specify arbitrary environment variables to be set for a task by providin Notice this example uses deep keys which can be more convenient but aren't as well supported by some toml implementations. +Declaring CLI options (experimental) +------------------------------------ + +By default extra CLI arguments are appended to the end of a cmd task, or exposed as +sys.argv in a script task. Alternatively it is possible to define CLI options that a +task should accept, which will be documented in the help for that task, and exposed to +the task in a way the makes the most sense for that task type. + +Arguments for cmd and shell tasks +~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ + +For cmd and shell tasks the values are exposed to the task as environment variables. For example given the following configuration: + +.. code-block:: toml + + [tool.poe.tasks.passby] + shell = """ + echo "hello $planet"; + echo "goodbye $planet"; + """ + help = "Pass by a planet!" + [tool.poe.tasks.passby.args.planet] # the key of the arg is used as the name of the variable that the given value will be exposed as + help = "Name of the planet to pass" + default = "earth" + required = false # by default all args are optional and default to "" + options = ["-p", "--planet"] # options are passed to ArgumentParser.add_argument as *args, if not given the the name value, i.e. [f"--{name}"] + +the resulting task can be run like: + +.. code-block:: bash + + poe passby --planet mars + +Arguments for script tasks +~~~~~~~~~~~~~~~~~~~~~~~~~~ + +Arguments can be defined for script tasks in the same way, but how they are exposed to +the underlying python function depends on how the script is defined. + +In the following example, since not parenthesis are included for the referenced function, +all provided args will be passed to the function as kwargs: + +.. code-block:: toml + + [tool.poe.tasks] + build = { script = "project.util:build", args = ["dest", "version"] + +Here the build method will be passed the two argument values (if provided) from the +command lines as kwargs. + +Note that in this example, args are given as a list of strings. This abbreviated +form is equivalent to just providing a name for each argument and keeping the default +values for all other configuration (including empty string for the help message). + +If there's a need to take control of how values are passed to the function, then this +is also possible as demonstrated in the following example: + +.. code-block:: toml + + [tool.poe.tasks] + build = { script = "project.util:build(dest, build_version=version)", args = ["dest", "version"] + Project-wide configuration options ================================== @@ -358,8 +420,6 @@ There's plenty to do, come say hi in the issues! 👋 TODO ==== -☐ support declaring specific arguments for a task `#6 `_ - ☐ support conditional execution (a bit like make targets) `#12 `_ ☐ support verbose mode for documentation that shows task definitions diff --git a/poethepoet/app.py b/poethepoet/app.py index b385c622b..df9c36079 100644 --- a/poethepoet/app.py +++ b/poethepoet/app.py @@ -1,11 +1,12 @@ import os from pathlib import Path import sys -from typing import Any, IO, MutableMapping, Optional, Sequence, Union +from typing import Any, Dict, IO, MutableMapping, Optional, Sequence, Tuple, Union from .config import PoeConfig from .context import RunContext from .exceptions import ExecutionError, PoeException from .task import PoeTask +from .task.args import PoeTaskArgs from .ui import PoeUi @@ -100,8 +101,15 @@ def print_help( ): if isinstance(error, str): error == PoeException(error) - tasks_help = { - task: (content.get("help", "") if isinstance(content, dict) else "") - for task, content in self.config.tasks.items() + tasks_help: Dict[str, Tuple[str, Sequence[Tuple[Tuple[str, ...], str]]]] = { + task_name: ( + ( + content.get("help", ""), + PoeTaskArgs.get_help_content(content.get("args")), + ) + if isinstance(content, dict) + else ("", tuple()) + ) + for task_name, content in self.config.tasks.items() } self.ui.print_help(tasks=tasks_help, info=info, error=error) # type: ignore diff --git a/poethepoet/task/args.py b/poethepoet/task/args.py new file mode 100644 index 000000000..f9bc502a3 --- /dev/null +++ b/poethepoet/task/args.py @@ -0,0 +1,154 @@ +import argparse +from typing import ( + Any, + Dict, + List, + Optional, + Sequence, + Set, + Tuple, + Type, + Union, + TYPE_CHECKING, +) + +if TYPE_CHECKING: + from ..config import PoeConfig + +ArgParams = Dict[str, Any] +ArgsDef = Union[List[str], List[ArgParams], Dict[str, ArgParams]] +arg_param_schema: Dict[str, Union[Type, Tuple[Type, ...]]] = { + "default": (str, int, float, bool), + "help": str, + "name": str, + "options": (list, tuple), + "required": bool, +} + + +class PoeTaskArgs: + _args: Tuple[ArgParams, ...] + + def __init__(self, args_def: ArgsDef): + self._args = self._normalize_args_def(args_def) + + @staticmethod + def _normalize_args_def(args_def: ArgsDef): + """ + args_def can be defined as a dictionary of ArgParams, or a list of strings, or + ArgParams. Here we normalize it to a list of ArgParams, assuming that it has + already been validated. + """ + result = [] + if isinstance(args_def, list): + for item in args_def: + if isinstance(item, str): + result.append({"name": item, "options": (f"--{item}",)}) + else: + result.append( + dict( + item, + options=tuple( + item.get("options", (f"--{item.get('name')}",)) + ), + ) + ) + else: + for name, params in args_def.items(): + result.append( + dict( + params, + name=name, + options=tuple(params.get("options", (f"--{name}",))), + ) + ) + return result + + @classmethod + def get_help_content( + cls, args_def: Optional[ArgsDef] + ) -> List[Tuple[Tuple[str, ...], str]]: + if args_def is None: + return [] + args = cls._normalize_args_def(args_def) + return [(arg["options"], arg.get("help", "")) for arg in args] + + @classmethod + def validate_def(cls, task_name: str, args_def: ArgsDef) -> Optional[str]: + arg_names: Set[str] = set() + if isinstance(args_def, list): + for item in args_def: + # can be a list of strings (just arg name) or ArgConfig dictionaries + if isinstance(item, str): + arg_name = item + elif isinstance(item, dict): + arg_name = item.get("name", "") + error = cls._validate_params(item, arg_name, task_name) + if error: + return error + else: + return f"Arg {item!r} of task {task_name!r} has invlaid type" + error = cls._validate_name(arg_name, task_name, arg_names) + if error: + return error + elif isinstance(args_def, dict): + for arg_name, params in args_def.items(): + error = cls._validate_name(arg_name, task_name, arg_names) + if error: + return error + if "name" in params: + return ( + f"Unexpected 'name' option for arg {arg_name!r} of task " + f"{task_name!r}" + ) + error = cls._validate_params(params, arg_name, task_name) + if error: + return error + return None + + @classmethod + def _validate_params( + cls, params: ArgParams, arg_name: str, task_name: str + ) -> Optional[str]: + for param, value in params.items(): + if param not in arg_param_schema: + return ( + f"Invalid option {param!r} for arg {arg_name!r} of task " + f"{task_name!r}" + ) + if not isinstance(value, arg_param_schema[param]): + return ( + f"Invalid value for option {param!r} of arg {arg_name!r} of" + f" task {task_name!r}" + ) + return None + + @classmethod + def _validate_name( + cls, name: Any, task_name: str, arg_names: Set[str] + ) -> Optional[str]: + if not isinstance(name, str): + return f"Arg name {name!r} of task {task_name!r} should be a string" + if not name.isidentifier(): + return ( + f"Arg name {name!r} of task {task_name!r} is not a valid " "identifier" + ) + if name in arg_names: + return f"Duplicate arg name {name!r} for task {task_name!r}" + arg_names.add(name) + return None + + def build_parser(self) -> argparse.ArgumentParser: + parser = argparse.ArgumentParser(add_help=False, allow_abbrev=False) + for arg in self._args: + parser.add_argument( + *arg["options"], + default=arg.get("default", ""), + dest=arg["name"], + required=arg.get("required", False), + help=arg.get("help", ""), + ) + return parser + + def parse(self, extra_args: Sequence[str]): + return vars(self.build_parser().parse_args(extra_args)) diff --git a/poethepoet/task/base.py b/poethepoet/task/base.py index 71644ec20..d8403c988 100644 --- a/poethepoet/task/base.py +++ b/poethepoet/task/base.py @@ -3,15 +3,16 @@ from typing import ( Any, Dict, - Iterable, List, MutableMapping, Optional, + Sequence, Tuple, Type, TYPE_CHECKING, Union, ) +from .args import PoeTaskArgs from ..exceptions import PoeException if TYPE_CHECKING: @@ -60,7 +61,12 @@ class PoeTask(metaclass=MetaPoeTask): __options__: Dict[str, Type] = {} __content_type__: Type = str - __base_options: Dict[str, Type] = {"env": dict, "executor": dict, "help": str} + __base_options: Dict[str, Union[Type, Tuple[Type, ...]]] = { + "args": (dict, list), + "env": dict, + "executor": dict, + "help": str, + } __task_types: Dict[str, Type["PoeTask"]] = {} def __init__( @@ -130,7 +136,7 @@ def from_def( def run( self, context: "RunContext", - extra_args: Iterable[str], + extra_args: Sequence[str], env: Optional[MutableMapping[str, str]] = None, ) -> int: """ @@ -141,10 +147,16 @@ def run( env = dict(env, **self.options["env"]) return self._handle_run(context, extra_args, env) + def parse_named_args(self, extra_args: Sequence[str]) -> Optional[Dict[str, str]]: + args_def = self.options.get("args") + if args_def: + return PoeTaskArgs(args_def).parse(extra_args) + return None + def _handle_run( self, context: "RunContext", - extra_args: Iterable[str], + extra_args: Sequence[str], env: MutableMapping[str, str], ) -> int: """ @@ -211,47 +223,51 @@ def validate_def( ) elif isinstance(task_def, dict): task_type_keys = set(task_def.keys()).intersection(cls.__task_types) - if len(task_type_keys) == 1: - task_type_key = next(iter(task_type_keys)) - task_content = task_def[task_type_key] - 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__}" + if len(task_type_keys) != 1: + return ( + f"Invalid task: {task_name!r}. Task definition must include exactly" + f" one task key from {set(cls.__task_types)!r}" + ) + task_type_key = next(iter(task_type_keys)) + task_content = task_def[task_type_key] + 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__}" + ) + 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: + 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: - for key in set(task_def) - {task_type_key}: - expected_type = cls.__base_options.get( - key, task_type.__options__.get(key) + if hasattr(task_type, "_validate_task_def"): + task_type_issue = task_type._validate_task_def( + task_name, task_def, config ) - if expected_type is None: - 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 - if "\n" in task_def.get("help", ""): - return ( - f"Invalid task: {task_name!r}. Help messages cannot contain " - "line breaks" - ) - else: + if task_type_issue: + return task_type_issue + + if "\n" in task_def.get("help", ""): return ( - f"Invalid task: {task_name!r}. Task definition must include exactly" - f" one task key from {set(cls.__task_types)!r}" + f"Invalid task: {task_name!r}. Help messages cannot contain " + "line breaks" ) + + if "args" in task_def: + return PoeTaskArgs.validate_def(task_name, task_def["args"]) + return None @classmethod diff --git a/poethepoet/task/cmd.py b/poethepoet/task/cmd.py index bce8e690b..62d1f1fc1 100644 --- a/poethepoet/task/cmd.py +++ b/poethepoet/task/cmd.py @@ -3,8 +3,8 @@ import shlex from typing import ( Dict, - Iterable, MutableMapping, + Sequence, Type, TYPE_CHECKING, ) @@ -31,13 +31,27 @@ class CmdTask(PoeTask): def _handle_run( self, context: "RunContext", - extra_args: Iterable[str], + extra_args: Sequence[str], env: MutableMapping[str, str], ) -> int: - cmd = (*self._resolve_args(context, env), *extra_args) + env, has_named_args = self._add_named_args_to_env(extra_args, env) + if has_named_args: + # If named arguments are defined then it doesn't make sense to pass extra + # args to the command, because they've already been parsed + cmd = self._resolve_args(context, env) + else: + cmd = (*self._resolve_args(context, env), *extra_args) self._print_action(" ".join(cmd), context.dry) return context.get_executor(env, self.options.get("executor")).execute(cmd) + def _add_named_args_to_env( + self, extra_args: Sequence[str], env: MutableMapping[str, str] + ): + named_args = self.parse_named_args(extra_args) + if named_args is None: + return env, False + return dict(env, **named_args), bool(named_args) + def _resolve_args( self, context: "RunContext", env: MutableMapping[str, str], ): diff --git a/poethepoet/task/ref.py b/poethepoet/task/ref.py index d6062e85a..5fcc9c992 100644 --- a/poethepoet/task/ref.py +++ b/poethepoet/task/ref.py @@ -1,9 +1,9 @@ from typing import ( Any, Dict, - Iterable, MutableMapping, Optional, + Sequence, Type, TYPE_CHECKING, ) @@ -29,7 +29,7 @@ class RefTask(PoeTask): def _handle_run( self, context: "RunContext", - extra_args: Iterable[str], + extra_args: Sequence[str], env: MutableMapping[str, str], ) -> int: """ diff --git a/poethepoet/task/script.py b/poethepoet/task/script.py index d2c03a58a..8f094569f 100644 --- a/poethepoet/task/script.py +++ b/poethepoet/task/script.py @@ -2,12 +2,12 @@ from typing import ( Any, Dict, - Iterable, Optional, MutableMapping, Tuple, Type, TYPE_CHECKING, + Sequence, Union, ) from .base import PoeTask @@ -33,12 +33,19 @@ class ScriptTask(PoeTask): def _handle_run( self, context: "RunContext", - extra_args: Iterable[str], + extra_args: Sequence[str], env: MutableMapping[str, str], ) -> 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, target_call = self._parse_content(self.content) + target_module, target_callable, call_params = self._parse_content(self.content) + named_args = self.parse_named_args(extra_args) + if named_args is not None: + if call_params is None: + call_params = f"**({named_args!r})" + else: + call_params = self._resolve_args(call_params, named_args) + argv = [ self.name, *(self._resolve_envvars(token, context, env) for token in extra_args), @@ -49,41 +56,56 @@ def _handle_run( "import sys; " "from importlib import import_module; " f"sys.argv = {argv!r}; sys.path.append('src');" - f"import_module('{target_module}').{target_call}", + f"import_module('{target_module}').{target_callable}({call_params or ''})", ) self._print_action(" ".join(argv), context.dry) return context.get_executor(env, self.options.get("executor")).execute(cmd) @classmethod - def _parse_content(cls, call_ref: str) -> Union[Tuple[str, str], Tuple[None, None]]: + def _parse_content( + cls, call_ref: str + ) -> Union[Tuple[str, str, Optional[str]], Tuple[None, None, None]]: """ - Parse module and callable call out of a string like one of: + Parse module and callable call from a string like one of: - "some_module:main" - "some.module:main(foo='bar')" """ try: target_module, target_ref = call_ref.split(":") except ValueError: - return None, None + return None, None, None if target_ref.isidentifier(): - return target_module, f"{target_ref}()" + return target_module, f"{target_ref}", None call_match = _FUNCTION_CALL_PATTERN.match(target_ref) if call_match: callable_name, call_params = call_match.groups() - return target_module, f"{callable_name}({call_params})" + return target_module, callable_name, call_params - return None, None + return None, None, None @classmethod def _validate_task_def( cls, task_name: str, task_def: Dict[str, Any], config: "PoeConfig" ) -> Optional[str]: - target_module, target_call = cls._parse_content(task_def["script"]) - if not target_module or not target_call: + target_module, target_callable, _ = cls._parse_content(task_def["script"]) + if not target_module or not target_callable: return ( f"Task {task_name!r} contains invalid callable reference " f"{task_def['script']!r} (expected something like `module:callable()`)" ) return None + + @classmethod + def _resolve_args(cls, call_params: str, named_args: Dict[str, str]): + def resolve_param(param: str): + if "=" in param: + keyword, value = (token.strip() for token in param.split(r"=", 1)) + return f"{keyword}={named_args.get(value, value)!r}" + else: + return named_args.get(param, param) + + return ", ".join( + resolve_param(param.strip()) for param in call_params.strip().split(",") + ) diff --git a/poethepoet/task/sequence.py b/poethepoet/task/sequence.py index 8b9d89a84..1de03f963 100644 --- a/poethepoet/task/sequence.py +++ b/poethepoet/task/sequence.py @@ -1,10 +1,10 @@ from typing import ( Any, Dict, - Iterable, List, MutableMapping, Optional, + Sequence, Type, TYPE_CHECKING, Union, @@ -53,7 +53,7 @@ def __init__( def _handle_run( self, context: "RunContext", - extra_args: Iterable[str], + extra_args: Sequence[str], env: MutableMapping[str, str], ) -> int: if any(arg.strip() for arg in extra_args): diff --git a/poethepoet/task/shell.py b/poethepoet/task/shell.py index 5c4f1c86c..ce03581d4 100644 --- a/poethepoet/task/shell.py +++ b/poethepoet/task/shell.py @@ -1,7 +1,7 @@ import os import shutil import subprocess -from typing import Dict, Iterable, MutableMapping, Type, TYPE_CHECKING +from typing import Dict, MutableMapping, Sequence, Type, TYPE_CHECKING from ..exceptions import PoeException from .base import PoeTask @@ -23,10 +23,12 @@ class ShellTask(PoeTask): def _handle_run( self, context: "RunContext", - extra_args: Iterable[str], + extra_args: Sequence[str], env: MutableMapping[str, str], ) -> int: - if any(arg.strip() for arg in extra_args): + env, has_named_args = self._add_named_args_to_env(extra_args, env) + + if not has_named_args and any(arg.strip() for arg in extra_args): raise PoeException(f"Shell task {self.name!r} does not accept arguments") if self._is_windows: @@ -40,6 +42,14 @@ def _handle_run( shell, input=self.content.encode() ) + def _add_named_args_to_env( + self, extra_args: Sequence[str], env: MutableMapping[str, str] + ): + named_args = self.parse_named_args(extra_args) + if named_args is None: + return env, False + return dict(env, **named_args), bool(named_args) + @staticmethod def _find_posix_shell_on_windows(): # Try locate a shell from the environment diff --git a/poethepoet/ui.py b/poethepoet/ui.py index a4e64d8df..8f46b0675 100644 --- a/poethepoet/ui.py +++ b/poethepoet/ui.py @@ -2,7 +2,7 @@ import os from pastel import Pastel import sys -from typing import IO, List, Mapping, Optional, Sequence, Union +from typing import IO, List, Mapping, Optional, Sequence, Tuple, Union from .exceptions import PoeException from .__version__ import __version__ @@ -32,6 +32,7 @@ def _init_colors(self): self._color.add_style("hl", "light_gray") self._color.add_style("em", "cyan") self._color.add_style("em2", "cyan", options="italic") + self._color.add_style("em3", "blue") self._color.add_style("h2", "default", options="bold") self._color.add_style("h2-dim", "default", options="dark") self._color.add_style("action", "light_blue") @@ -41,7 +42,7 @@ def __getitem__(self, key: str): """Provide easy access to arguments""" return getattr(self.args, key, None) - def build_parser(self): + def build_parser(self) -> argparse.ArgumentParser: parser = argparse.ArgumentParser( prog="poe", description="Poe the Poet: A task runner that works well with poetry.", @@ -132,7 +133,7 @@ def parse_args(self, cli_args: Sequence[str]): def print_help( self, - tasks: Optional[Mapping[str, str]] = None, + tasks: Optional[Mapping[str, Tuple[str, Sequence[Tuple[str, str]]]]] = None, info: Optional[str] = None, error: Optional[PoeException] = None, ): @@ -179,15 +180,27 @@ def print_help( ) if tasks: - max_task_len = max(len(task) for task in tasks) + max_task_len = max( + max( + len(task), + max([len(", ".join(opts)) for (opts, _) in args] or (0,)) + 2, + ) + for task, (_, args) in tasks.items() + ) col_width = max(13, min(30, max_task_len)) tasks_section = ["

CONFIGURED TASKS

"] - for task, help_text in tasks.items(): + for task, (help_text, args_help) in tasks.items(): if task.startswith("_"): continue tasks_section.append( f" {self._padr(task, col_width)} {help_text}" ) + for (options, arg_help_text) in args_help: + tasks_section.append( + " " + f"{self._padr(', '.join(options), col_width - 2)}" + f" {arg_help_text}" + ) result.append(tasks_section) else: result.append("NO TASKS CONFIGURED")