Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/invoke/tasks.py: 21%
Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1"""
2This module contains the core `.Task` class & convenience decorators used to
3generate new tasks.
4"""
6import inspect
7import types
8from copy import deepcopy
9from functools import update_wrapper
10from typing import (
11 TYPE_CHECKING,
12 Any,
13 Callable,
14 Dict,
15 Generic,
16 Iterable,
17 List,
18 Optional,
19 Set,
20 Tuple,
21 Type,
22 TypeVar,
23 Union,
24)
26from .context import Context
27from .parser import Argument, ParseResult, translate_underscores
29if TYPE_CHECKING:
30 from inspect import Signature
32 from .config import Config
34T = TypeVar("T", bound=Callable)
37class Task(Generic[T]):
38 """
39 Core object representing an executable task & its argument specification.
41 For the most part, this object is a clearinghouse for all of the data that
42 may be supplied to the `@task <invoke.tasks.task>` decorator, such as
43 ``name``, ``aliases``, ``positional`` etc, which appear as attributes.
45 In addition, instantiation copies some introspection/documentation friendly
46 metadata off of the supplied ``body`` object, such as ``__doc__``,
47 ``__name__`` and ``__module__``, allowing it to "appear as" ``body`` for
48 most intents and purposes.
50 .. versionadded:: 1.0
51 """
53 # TODO: store these kwarg defaults central, refer to those values both here
54 # and in @task.
55 # TODO: allow central per-session / per-taskmodule control over some of
56 # them, e.g. (auto_)positional, auto_shortflags.
57 # NOTE: we shadow __builtins__.help here on purpose - obfuscating to avoid
58 # it feels bad, given the builtin will never actually be in play anywhere
59 # except a debug shell whose frame is exactly inside this class.
60 def __init__(
61 self,
62 body: Callable,
63 name: Optional[str] = None,
64 aliases: Iterable[str] = (),
65 positional: Optional[Iterable[str]] = None,
66 optional: Iterable[str] = (),
67 default: bool = False,
68 auto_shortflags: bool = True,
69 help: Optional[Dict[str, Any]] = None,
70 pre: Optional[Union[List[str], str]] = None,
71 post: Optional[Union[List[str], str]] = None,
72 autoprint: bool = False,
73 iterable: Optional[Iterable[str]] = None,
74 incrementable: Optional[Iterable[str]] = None,
75 ) -> None:
76 # Real callable
77 self.body = body
78 update_wrapper(self, self.body)
79 # Copy a bunch of special properties from the body for the benefit of
80 # Sphinx autodoc or other introspectors.
81 self.__doc__ = getattr(body, "__doc__", "")
82 self.__name__ = getattr(body, "__name__", "")
83 self.__module__ = getattr(body, "__module__", "")
84 # Default name, alternate names, and whether it should act as the
85 # default for its parent collection
86 self._name = name
87 self.aliases = aliases
88 self.is_default = default
89 # Arg/flag/parser hints
90 self.positional = self.fill_implicit_positionals(positional)
91 self.optional = tuple(optional)
92 self.iterable = iterable or []
93 self.incrementable = incrementable or []
94 self.auto_shortflags = auto_shortflags
95 self.help = (help or {}).copy()
96 # Call chain bidness
97 self.pre = pre or []
98 self.post = post or []
99 self.times_called = 0
100 # Whether to print return value post-execution
101 self.autoprint = autoprint
103 @property
104 def name(self) -> str:
105 return self._name or self.__name__
107 def __repr__(self) -> str:
108 aliases = ""
109 if self.aliases:
110 aliases = " ({})".format(", ".join(self.aliases))
111 return "<Task {!r}{}>".format(self.name, aliases)
113 def __eq__(self, other: object) -> bool:
114 if not isinstance(other, Task) or self.name != other.name:
115 return False
116 # Functions do not define __eq__ but func_code objects apparently do.
117 # (If we're wrapping some other callable, they will be responsible for
118 # defining equality on their end.)
119 if self.body == other.body:
120 return True
121 else:
122 try:
123 return self.body.__code__ == other.body.__code__
124 except AttributeError:
125 return False
127 def __hash__(self) -> int:
128 # Presumes name and body will never be changed. Hrm.
129 # Potentially cleaner to just not use Tasks as hash keys, but let's do
130 # this for now.
131 return hash(self.name) + hash(self.body)
133 def __call__(self, *args: Any, **kwargs: Any) -> T:
134 # Guard against calling tasks with no context.
135 if not isinstance(args[0], Context):
136 err = "Task expected a Context as its first arg, got {} instead!"
137 # TODO: raise a custom subclass _of_ TypeError instead
138 raise TypeError(err.format(type(args[0])))
139 result = self.body(*args, **kwargs)
140 self.times_called += 1
141 return result
143 @property
144 def called(self) -> bool:
145 return self.times_called > 0
147 def argspec(self, body: Callable) -> "Signature":
148 """
149 Returns a modified `inspect.Signature` based on that of ``body``.
151 :returns:
152 an `inspect.Signature` matching that of ``body``, but with the
153 initial context argument removed.
154 :raises TypeError:
155 if the task lacks an initial positional `.Context` argument.
157 .. versionadded:: 1.0
158 .. versionchanged:: 2.0
159 Changed from returning a two-tuple of ``(arg_names, spec_dict)`` to
160 returning an `inspect.Signature`.
161 """
162 # Handle callable-but-not-function objects
163 if isinstance(body, types.FunctionType):
164 func = body
165 else:
166 func = body.__call__ # type: ignore
167 # Rebuild signature with first arg dropped, or die usefully(ish trying
168 sig = inspect.signature(func)
169 params = list(sig.parameters.values())
170 # TODO: this ought to also check if an extant 1st param _was_ a Context
171 # arg, and yell similarly if not.
172 if not len(params):
173 # TODO: see TODO under __call__, this should be same type
174 raise TypeError("Tasks must have an initial Context argument!")
175 return sig.replace(parameters=params[1:])
177 def fill_implicit_positionals(
178 self, positional: Optional[Iterable[str]]
179 ) -> Iterable[str]:
180 # If positionals is None, everything lacking a default
181 # value will be automatically considered positional.
182 if positional is None:
183 positional = [
184 x.name
185 for x in self.argspec(self.body).parameters.values()
186 if x.default is inspect.Signature.empty
187 ]
188 return positional
190 def arg_opts(
191 self, name: str, default: str, taken_names: Set[str]
192 ) -> Dict[str, Any]:
193 opts: Dict[str, Any] = {}
194 # Whether it's positional or not
195 opts["positional"] = name in self.positional
196 # Whether it is a value-optional flag
197 opts["optional"] = name in self.optional
198 # Whether it should be of an iterable (list) kind
199 if name in self.iterable:
200 opts["kind"] = list
201 # If user gave a non-None default, hopefully they know better
202 # than us what they want here (and hopefully it offers the list
203 # protocol...) - otherwise supply useful default
204 opts["default"] = default if default is not None else []
205 # Whether it should increment its value or not
206 if name in self.incrementable:
207 opts["incrementable"] = True
208 # Argument name(s) (replace w/ dashed version if underscores present,
209 # and move the underscored version to be the attr_name instead.)
210 original_name = name # For reference in eg help=
211 if "_" in name:
212 opts["attr_name"] = name
213 name = translate_underscores(name)
214 names = [name]
215 if self.auto_shortflags:
216 # Must know what short names are available
217 for char in name:
218 if not (char == name or char in taken_names):
219 names.append(char)
220 break
221 opts["names"] = names
222 # Handle default value & kind if possible
223 if default not in (None, inspect.Signature.empty):
224 # TODO: allow setting 'kind' explicitly.
225 # NOTE: skip setting 'kind' if optional is True + type(default) is
226 # bool; that results in a nonsensical Argument which gives the
227 # parser grief in a few ways.
228 kind = type(default)
229 if not (opts["optional"] and kind is bool):
230 opts["kind"] = kind
231 opts["default"] = default
232 # Help
233 for possibility in name, original_name:
234 if possibility in self.help:
235 opts["help"] = self.help.pop(possibility)
236 break
237 return opts
239 def get_arguments(
240 self, ignore_unknown_help: Optional[bool] = None
241 ) -> List[Argument]:
242 """
243 Return a list of Argument objects representing this task's signature.
245 :param bool ignore_unknown_help:
246 Controls whether unknown help flags cause errors. See the config
247 option by the same name for details.
249 .. versionadded:: 1.0
250 .. versionchanged:: 1.7
251 Added the ``ignore_unknown_help`` kwarg.
252 """
253 # Core argspec
254 sig = self.argspec(self.body)
255 # Prime the list of all already-taken names (mostly for help in
256 # choosing auto shortflags)
257 taken_names = set(sig.parameters.keys())
258 # Build arg list (arg_opts will take care of setting up shortnames,
259 # etc)
260 args = []
261 for param in sig.parameters.values():
262 new_arg = Argument(
263 **self.arg_opts(param.name, param.default, taken_names)
264 )
265 args.append(new_arg)
266 # Update taken_names list with new argument's full name list
267 # (which may include new shortflags) so subsequent Argument
268 # creation knows what's taken.
269 taken_names.update(set(new_arg.names))
270 # If any values were leftover after consuming a 'help' dict, it implies
271 # the user messed up & had a typo or similar. Let's explode.
272 if self.help and not ignore_unknown_help:
273 raise ValueError(
274 "Help field was set for param(s) that don't exist: {}".format(
275 list(self.help.keys())
276 )
277 )
278 # Now we need to ensure positionals end up in the front of the list, in
279 # order given in self.positionals, so that when Context consumes them,
280 # this order is preserved.
281 for posarg in reversed(list(self.positional)):
282 for i, arg in enumerate(args):
283 if arg.name == posarg:
284 args.insert(0, args.pop(i))
285 break
286 return args
289def task(*args: Any, **kwargs: Any) -> Callable:
290 """
291 Marks wrapped callable object as a valid Invoke task.
293 May be called without any parentheses if no extra options need to be
294 specified. Otherwise, the following keyword arguments are allowed in the
295 parenthese'd form:
297 * ``name``: Default name to use when binding to a `.Collection`. Useful for
298 avoiding Python namespace issues (i.e. when the desired CLI level name
299 can't or shouldn't be used as the Python level name.)
300 * ``aliases``: Specify one or more aliases for this task, allowing it to be
301 invoked as multiple different names. For example, a task named ``mytask``
302 with a simple ``@task`` wrapper may only be invoked as ``"mytask"``.
303 Changing the decorator to be ``@task(aliases=['myothertask'])`` allows
304 invocation as ``"mytask"`` *or* ``"myothertask"``.
305 * ``positional``: Iterable overriding the parser's automatic "args with no
306 default value are considered positional" behavior. If a list of arg
307 names, no args besides those named in this iterable will be considered
308 positional. (This means that an empty list will force all arguments to be
309 given as explicit flags.)
310 * ``optional``: Iterable of argument names, declaring those args to
311 have :ref:`optional values <optional-values>`. Such arguments may be
312 given as value-taking options (e.g. ``--my-arg=myvalue``, wherein the
313 task is given ``"myvalue"``) or as Boolean flags (``--my-arg``, resulting
314 in ``True``).
315 * ``iterable``: Iterable of argument names, declaring them to :ref:`build
316 iterable values <iterable-flag-values>`.
317 * ``incrementable``: Iterable of argument names, declaring them to
318 :ref:`increment their values <incrementable-flag-values>`.
319 * ``default``: Boolean option specifying whether this task should be its
320 collection's default task (i.e. called if the collection's own name is
321 given.)
322 * ``auto_shortflags``: Whether or not to automatically create short
323 flags from task options; defaults to True.
324 * ``help``: Dict mapping argument names to their help strings. Will be
325 displayed in ``--help`` output. For arguments containing underscores
326 (which are transformed into dashes on the CLI by default), either the
327 dashed or underscored version may be supplied here.
328 * ``pre``, ``post``: Lists of task objects to execute prior to, or after,
329 the wrapped task whenever it is executed.
330 * ``autoprint``: Boolean determining whether to automatically print this
331 task's return value to standard output when invoked directly via the CLI.
332 Defaults to False.
333 * ``klass``: Class to instantiate/return. Defaults to `.Task`.
335 If any non-keyword arguments are given, they are taken as the value of the
336 ``pre`` kwarg for convenience's sake. (It is an error to give both
337 ``*args`` and ``pre`` at the same time.)
339 .. versionadded:: 1.0
340 .. versionchanged:: 1.1
341 Added the ``klass`` keyword argument.
342 """
343 klass: Type[Task] = kwargs.pop("klass", Task)
344 # @task -- no options were (probably) given.
345 if len(args) == 1 and callable(args[0]) and not isinstance(args[0], Task):
346 return klass(args[0], **kwargs)
347 # @task(pre, tasks, here)
348 if args:
349 if "pre" in kwargs:
350 raise TypeError(
351 "May not give *args and 'pre' kwarg simultaneously!"
352 )
353 kwargs["pre"] = args
355 def inner(body: Callable) -> Task[T]:
356 _task = klass(body, **kwargs)
357 return _task
359 # update_wrapper(inner, klass)
360 return inner
363class Call:
364 """
365 Represents a call/execution of a `.Task` with given (kw)args.
367 Similar to `~functools.partial` with some added functionality (such as the
368 delegation to the inner task, and optional tracking of the name it's being
369 called by.)
371 .. versionadded:: 1.0
372 """
374 def __init__(
375 self,
376 task: "Task",
377 called_as: Optional[str] = None,
378 args: Optional[Tuple[str, ...]] = None,
379 kwargs: Optional[Dict[str, Any]] = None,
380 ) -> None:
381 """
382 Create a new `.Call` object.
384 :param task: The `.Task` object to be executed.
386 :param str called_as:
387 The name the task is being called as, e.g. if it was called by an
388 alias or other rebinding. Defaults to ``None``, aka, the task was
389 referred to by its default name.
391 :param tuple args:
392 Positional arguments to call with, if any. Default: ``None``.
394 :param dict kwargs:
395 Keyword arguments to call with, if any. Default: ``None``.
396 """
397 self.task = task
398 self.called_as = called_as
399 self.args = args or tuple()
400 self.kwargs = kwargs or dict()
402 # TODO: just how useful is this? feels like maybe overkill magic
403 def __getattr__(self, name: str) -> Any:
404 return getattr(self.task, name)
406 def __deepcopy__(self, memo: object) -> "Call":
407 return self.clone()
409 def __repr__(self) -> str:
410 aka = ""
411 if self.called_as is not None and self.called_as != self.task.name:
412 aka = " (called as: {!r})".format(self.called_as)
413 return "<{} {!r}{}, args: {!r}, kwargs: {!r}>".format(
414 self.__class__.__name__,
415 self.task.name,
416 aka,
417 self.args,
418 self.kwargs,
419 )
421 def __eq__(self, other: object) -> bool:
422 # NOTE: Not comparing 'called_as'; a named call of a given Task with
423 # same args/kwargs should be considered same as an unnamed call of the
424 # same Task with the same args/kwargs (e.g. pre/post task specified w/o
425 # name). Ditto tasks with multiple aliases.
426 for attr in "task args kwargs".split():
427 if getattr(self, attr) != getattr(other, attr):
428 return False
429 return True
431 def make_context(
432 self,
433 config: "Config",
434 core_parse_result: "ParseResult",
435 ) -> Context:
436 """
437 Generate a `.Context` appropriate for this call, with given config.
439 .. versionadded:: 1.0
440 .. versionchanged:: 3.0
441 Added the ``core_parse_result`` parameter.
442 """
443 return Context(config=config, remainder=core_parse_result.remainder)
445 def clone_data(self) -> Dict[str, Any]:
446 """
447 Return keyword args suitable for cloning this call into another.
449 .. versionadded:: 1.1
450 """
451 return dict(
452 task=self.task,
453 called_as=self.called_as,
454 args=deepcopy(self.args),
455 kwargs=deepcopy(self.kwargs),
456 )
458 def clone(
459 self,
460 into: Optional[Type["Call"]] = None,
461 with_: Optional[Dict[str, Any]] = None,
462 ) -> "Call":
463 """
464 Return a standalone copy of this Call.
466 Useful when parameterizing task executions.
468 :param into:
469 A subclass to generate instead of the current class. Optional.
471 :param dict with_:
472 A dict of additional keyword arguments to use when creating the new
473 clone; typically used when cloning ``into`` a subclass that has
474 extra args on top of the base class. Optional.
476 .. note::
477 This dict is used to ``.update()`` the original object's data
478 (the return value from its `clone_data`), so in the event of
479 a conflict, values in ``with_`` will win out.
481 .. versionadded:: 1.0
482 .. versionchanged:: 1.1
483 Added the ``with_`` kwarg.
484 """
485 klass = into if into is not None else self.__class__
486 data = self.clone_data()
487 if with_ is not None:
488 data.update(with_)
489 return klass(**data)
492def call(task: "Task", *args: Any, **kwargs: Any) -> "Call":
493 """
494 Describes execution of a `.Task`, typically with pre-supplied arguments.
496 Useful for setting up :ref:`pre/post task invocations
497 <parameterizing-pre-post-tasks>`. It's actually just a convenient wrapper
498 around the `.Call` class, which may be used directly instead if desired.
500 For example, here's two build-like tasks that both refer to a ``setup``
501 pre-task, one with no baked-in argument values (and thus no need to use
502 `.call`), and one that toggles a boolean flag::
504 @task
505 def setup(c, clean=False):
506 if clean:
507 c.run("rm -rf target")
508 # ... setup things here ...
509 c.run("tar czvf target.tgz target")
511 @task(pre=[setup])
512 def build(c):
513 c.run("build, accounting for leftover files...")
515 @task(pre=[call(setup, clean=True)])
516 def clean_build(c):
517 c.run("build, assuming clean slate...")
519 Please see the constructor docs for `.Call` for details - this function's
520 ``args`` and ``kwargs`` map directly to the same arguments as in that
521 method.
523 .. versionadded:: 1.0
524 """
525 return Call(task, args=args, kwargs=kwargs)