/usr/lib/python3/dist-packages/invoke/tasks.py is in python3-invoke 0.11.1+dfsg1-1.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 | """
This module contains the core `.Task` class & convenience decorators used to
generate new tasks.
"""
import inspect
import types
import six
from .context import Context
from .parser import Argument
if six.PY3:
from itertools import zip_longest
else:
from itertools import izip_longest as zip_longest
#: Sentinel object representing a truly blank value (vs ``None``).
NO_DEFAULT = object()
class Task(object):
"""
Core object representing an executable task & its argument specification.
"""
# TODO: store these kwarg defaults central, refer to those values both here
# and in @task.
# TODO: allow central per-session / per-taskmodule control over some of
# them, e.g. (auto_)positional, auto_shortflags.
# NOTE: we shadow __builtins__.help here. It's purposeful. :(
def __init__(self,
body,
name=None,
contextualized=False,
aliases=(),
positional=None,
optional=(),
default=False,
auto_shortflags=True,
help=None,
pre=None,
post=None,
autoprint=False,
):
# Real callable
self.body = body
# Must copy doc/name here because Sphinx is stupid about properties.
self.__doc__ = getattr(body, '__doc__', '')
self.__name__ = getattr(body, '__name__', '')
# Is this a contextualized task?
self.contextualized = contextualized
# Default name, alternate names, and whether it should act as the
# default for its parent collection
self._name = name
self.aliases = aliases
self.is_default = default
# Arg/flag/parser hints
self.positional = self.fill_implicit_positionals(positional)
self.optional = optional
self.auto_shortflags = auto_shortflags
self.help = help or {}
# Call chain bidness
self.pre = pre or []
self.post = post or []
self.times_called = 0
# Whether to print return value post-execution
self.autoprint = autoprint
@property
def name(self):
return self._name or self.__name__
def __str__(self):
aliases = ""
if self.aliases:
aliases = " ({0})".format(', '.join(self.aliases))
return "<Task {0!r}{1}>".format(self.name, aliases)
def __repr__(self):
return str(self)
def __eq__(self, other):
if self.name != other.name:
return False
# Functions do not define __eq__ but func_code objects apparently do.
# (If we're wrapping some other callable, they will be responsible for
# defining equality on their end.)
if self.body == other.body:
return True
else:
try:
return (
six.get_function_code(self.body) ==
six.get_function_code(other.body)
)
except AttributeError:
return False
def __hash__(self):
# Presumes name and body will never be changed. Hrm.
# Potentially cleaner to just not use Tasks as hash keys, but let's do
# this for now.
return hash(self.name) + hash(self.body)
def __call__(self, *args, **kwargs):
# Guard against calling contextualized tasks with no context.
if self.contextualized and not isinstance(args[0], Context):
err = "Contextualized task expected a Context, got {0} instead!"
raise TypeError(err.format(type(args[0])))
result = self.body(*args, **kwargs)
self.times_called += 1
return result
@property
def called(self):
return self.times_called > 0
def argspec(self, body):
"""
Returns two-tuple:
* First item is list of arg names, in order defined.
* I.e. we *cannot* simply use a dict's ``keys()`` method here.
* Second item is dict mapping arg names to default values or
`.NO_DEFAULT` (an 'empty' value distinct from None, since None
is a valid value on its own).
"""
# Handle callable-but-not-function objects
# TODO: __call__ exhibits the 'self' arg; do we manually nix 1st result
# in argspec, or is there a way to get the "really callable" spec?
func = body if isinstance(body, types.FunctionType) else body.__call__
spec = inspect.getargspec(func)
arg_names = spec.args[:]
matched_args = [reversed(x) for x in [spec.args, spec.defaults or []]]
spec_dict = dict(zip_longest(*matched_args, fillvalue=NO_DEFAULT))
# Remove context argument, if applicable
if self.contextualized:
context_arg = arg_names.pop(0)
del spec_dict[context_arg]
return arg_names, spec_dict
def fill_implicit_positionals(self, positional):
args, spec_dict = self.argspec(self.body)
# If positionals is None, everything lacking a default
# value will be automatically considered positional.
if positional is None:
positional = []
for name in args: # Go in defined order, not dict "order"
default = spec_dict[name]
if default is NO_DEFAULT:
positional.append(name)
return positional
def arg_opts(self, name, default, taken_names):
opts = {}
# Whether it's positional or not
opts['positional'] = name in self.positional
# Whether it is a value-optional flag
opts['optional'] = name in self.optional
# Argument name(s) (replace w/ dashed version if underscores present,
# and move the underscored version to be the attr_name instead.)
if '_' in name:
opts['attr_name'] = name
name = name.replace('_', '-')
names = [name]
if self.auto_shortflags:
# Must know what short names are available
for char in name:
if not (char == name or char in taken_names):
names.append(char)
break
opts['names'] = names
# Handle default value & kind if possible
if default not in (None, NO_DEFAULT):
# TODO: allow setting 'kind' explicitly.
opts['kind'] = type(default)
opts['default'] = default
# Help
if name in self.help:
opts['help'] = self.help[name]
return opts
def get_arguments(self):
"""
Return a list of Argument objects representing this task's signature.
"""
# Core argspec
arg_names, spec_dict = self.argspec(self.body)
# Obtain list of args + their default values (if any) in
# declaration/definition order (i.e. based on getargspec())
tuples = [(x, spec_dict[x]) for x in arg_names]
# Prime the list of all already-taken names (mostly for help in
# choosing auto shortflags)
taken_names = set(x[0] for x in tuples)
# Build arg list (arg_opts will take care of setting up shortnames,
# etc)
args = []
for name, default in tuples:
new_arg = Argument(**self.arg_opts(name, default, taken_names))
args.append(new_arg)
# Update taken_names list with new argument's full name list
# (which may include new shortflags) so subsequent Argument
# creation knows what's taken.
taken_names.update(set(new_arg.names))
# Now we need to ensure positionals end up in the front of the list, in
# order given in self.positionals, so that when Context consumes them,
# this order is preserved.
for posarg in reversed(self.positional):
for i, arg in enumerate(args):
if arg.name == posarg:
args.insert(0, args.pop(i))
break
return args
def task(*args, **kwargs):
"""
Marks wrapped callable object as a valid Invoke task.
May be called without any parentheses if no extra options need to be
specified. Otherwise, the following keyword arguments are allowed in the
parenthese'd form:
* ``name``: Default name to use when binding to a `.Collection`. Useful for
avoiding Python namespace issues (i.e. when the desired CLI level name
can't or shouldn't be used as the Python level name.)
* ``contextualized``: Hints to callers (especially the CLI) that this task
expects to be given a `.Context` object as its first argument when
called.
* ``aliases``: Specify one or more aliases for this task, allowing it to be
invoked as multiple different names. For example, a task named ``mytask``
with a simple ``@task`` wrapper may only be invoked as ``"mytask"``.
Changing the decorator to be ``@task(aliases=['myothertask'])`` allows
invocation as ``"mytask"`` *or* ``"myothertask"``.
* ``positional``: Iterable overriding the parser's automatic "args with no
default value are considered positional" behavior. If a list of arg
names, no args besides those named in this iterable will be considered
positional. (This means that an empty list will force all arguments to be
given as explicit flags.)
* ``optional``: Iterable of argument names, declaring those args to
have :ref:`optional values <optional-values>`. Such arguments may be
given as value-taking options (e.g. ``--my-arg=myvalue``, wherein the
task is given ``"myvalue"``) or as Boolean flags (``--my-arg``, resulting
in ``True``).
* ``default``: Boolean option specifying whether this task should be its
collection's default task (i.e. called if the collection's own name is
given.)
* ``auto_shortflags``: Whether or not to automatically create short
flags from task options; defaults to True.
* ``help``: Dict mapping argument names to their help strings. Will be
displayed in ``--help`` output.
* ``pre``, ``post``: Lists of task objects to execute prior to, or after,
the wrapped task whenever it is executed.
* ``autoprint``: Boolean determining whether to automatically print this
task's return value to standard output when invoked directly via the CLI.
Defaults to False.
If any non-keyword arguments are given, they are taken as the value of the
``pre`` kwarg for convenience's sake. (It is an error to give both
``*args`` and ``pre`` at the same time.)
"""
# @task -- no options were (probably) given.
# Also handles ctask's use case when given as @ctask, equivalent to
# @task(obj, contextualized=True).
if len(args) == 1 and callable(args[0]) and not isinstance(args[0], Task):
return Task(args[0], **kwargs)
# @task(pre, tasks, here)
if args:
if 'pre' in kwargs:
raise TypeError(
"May not give *args and 'pre' kwarg simultaneously!"
)
kwargs['pre'] = args
# @task(options)
# TODO: pull in centrally defined defaults here (see Task)
name = kwargs.pop('name', None)
contextualized = kwargs.pop('contextualized', False)
aliases = kwargs.pop('aliases', ())
positional = kwargs.pop('positional', None)
optional = tuple(kwargs.pop('optional', ()))
default = kwargs.pop('default', False)
auto_shortflags = kwargs.pop('auto_shortflags', True)
help = kwargs.pop('help', {})
pre = kwargs.pop('pre', [])
post = kwargs.pop('post', [])
autoprint = kwargs.pop('autoprint', False)
# Handle unknown kwargs
if kwargs:
kwarg = (" unknown kwargs {0!r}".format(kwargs)) if kwargs else ""
raise TypeError("@task was called with" + kwarg)
def inner(obj):
obj = Task(
obj,
name=name,
contextualized=contextualized,
aliases=aliases,
positional=positional,
optional=optional,
default=default,
auto_shortflags=auto_shortflags,
help=help,
pre=pre,
post=post,
autoprint=autoprint,
)
return obj
return inner
def ctask(*args, **kwargs):
"""
Wrapper for `.task` which sets ``contextualized=True`` by default.
Please see `.task` for documentation.
"""
kwargs.setdefault('contextualized', True)
return task(*args, **kwargs)
class Call(object):
"""
Represents a call/execution of a `.Task` with some arguments.
Wraps its `.Task` so it can be treated as one by `.Executor`.
Similar to `~functools.partial` with some added functionality (such as the
delegation to the inner task, and optional tracking of a given name.)
"""
def __init__(self, task, *args, **kwargs):
self.task = task
self.args = args
self.kwargs = kwargs
self.name = None # Mostly manipulated by client code
def __getattr__(self, name):
return getattr(self.task, name)
def __str__(self):
return "<Call {0!r}, args: {1!r} kwargs: {2!r}>".format(
self.task.name, self.args, self.kwargs
)
def __repr__(self):
return str(self)
# Convenience/aesthetically pleasing-ish alias
call = Call
|