Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/invoke/collection.py: 17%
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
1import copy
2from types import ModuleType
3from typing import Any, Callable, Dict, List, Optional, Tuple
5from .util import Lexicon, helpline
7from .config import merge_dicts, copy_dict
8from .parser import Context as ParserContext
9from .tasks import Task
12class Collection:
13 """
14 A collection of executable tasks. See :doc:`/concepts/namespaces`.
16 .. versionadded:: 1.0
17 """
19 def __init__(self, *args: Any, **kwargs: Any) -> None:
20 """
21 Create a new task collection/namespace.
23 `.Collection` offers a set of methods for building a collection of
24 tasks from scratch, plus a convenient constructor wrapping said API.
26 In either case:
28 * The first positional argument may be a string, which (if given) is
29 used as the collection's default name when performing namespace
30 lookups;
31 * A ``loaded_from`` keyword argument may be given, which sets metadata
32 indicating the filesystem path the collection was loaded from. This
33 is used as a guide when loading per-project :ref:`configuration files
34 <config-hierarchy>`.
35 * An ``auto_dash_names`` kwarg may be given, controlling whether task
36 and collection names have underscores turned to dashes in most cases;
37 it defaults to ``True`` but may be set to ``False`` to disable.
39 The CLI machinery will pass in the value of the
40 ``tasks.auto_dash_names`` config value to this kwarg.
42 **The method approach**
44 May initialize with no arguments and use methods (e.g.
45 `.add_task`/`.add_collection`) to insert objects::
47 c = Collection()
48 c.add_task(some_task)
50 If an initial string argument is given, it is used as the default name
51 for this collection, should it be inserted into another collection as a
52 sub-namespace::
54 docs = Collection('docs')
55 docs.add_task(doc_task)
56 ns = Collection()
57 ns.add_task(top_level_task)
58 ns.add_collection(docs)
59 # Valid identifiers are now 'top_level_task' and 'docs.doc_task'
60 # (assuming the task objects were actually named the same as the
61 # variables we're using :))
63 For details, see the API docs for the rest of the class.
65 **The constructor approach**
67 All ``*args`` given to `.Collection` (besides the abovementioned
68 optional positional 'name' argument and ``loaded_from`` kwarg) are
69 expected to be `.Task` or `.Collection` instances which will be passed
70 to `.add_task`/`.add_collection` as appropriate. Module objects are
71 also valid (as they are for `.add_collection`). For example, the below
72 snippet results in the same two task identifiers as the one above::
74 ns = Collection(top_level_task, Collection('docs', doc_task))
76 If any ``**kwargs`` are given, the keywords are used as the initial
77 name arguments for the respective values::
79 ns = Collection(
80 top_level_task=some_other_task,
81 docs=Collection(doc_task)
82 )
84 That's exactly equivalent to::
86 docs = Collection(doc_task)
87 ns = Collection()
88 ns.add_task(some_other_task, 'top_level_task')
89 ns.add_collection(docs, 'docs')
91 See individual methods' API docs for details.
92 """
93 # Initialize
94 self.tasks = Lexicon()
95 self.collections = Lexicon()
96 self.default: Optional[str] = None
97 self.name = None
98 self._configuration: Dict[str, Any] = {}
99 # Specific kwargs if applicable
100 self.loaded_from = kwargs.pop("loaded_from", None)
101 self.auto_dash_names = kwargs.pop("auto_dash_names", None)
102 # splat-kwargs version of default value (auto_dash_names=True)
103 if self.auto_dash_names is None:
104 self.auto_dash_names = True
105 # Name if applicable
106 _args = list(args)
107 if _args and isinstance(args[0], str):
108 self.name = self.transform(_args.pop(0))
109 # Dispatch args/kwargs
110 for arg in _args:
111 self._add_object(arg)
112 # Dispatch kwargs
113 for name, obj in kwargs.items():
114 self._add_object(obj, name)
116 def _add_object(self, obj: Any, name: Optional[str] = None) -> None:
117 method: Callable
118 if isinstance(obj, Task):
119 method = self.add_task
120 elif isinstance(obj, (Collection, ModuleType)):
121 method = self.add_collection
122 else:
123 raise TypeError("No idea how to insert {!r}!".format(type(obj)))
124 method(obj, name=name)
126 def __repr__(self) -> str:
127 task_names = list(self.tasks.keys())
128 collections = ["{}...".format(x) for x in self.collections.keys()]
129 return "<Collection {!r}: {}>".format(
130 self.name, ", ".join(sorted(task_names) + sorted(collections))
131 )
133 def __eq__(self, other: object) -> bool:
134 if isinstance(other, Collection):
135 return (
136 self.name == other.name
137 and self.tasks == other.tasks
138 and self.collections == other.collections
139 )
140 return False
142 def __bool__(self) -> bool:
143 return bool(self.task_names)
145 @classmethod
146 def from_module(
147 cls,
148 module: ModuleType,
149 name: Optional[str] = None,
150 config: Optional[Dict[str, Any]] = None,
151 loaded_from: Optional[str] = None,
152 auto_dash_names: Optional[bool] = None,
153 ) -> "Collection":
154 """
155 Return a new `.Collection` created from ``module``.
157 Inspects ``module`` for any `.Task` instances and adds them to a new
158 `.Collection`, returning it. If any explicit namespace collections
159 exist (named ``ns`` or ``namespace``) a copy of that collection object
160 is preferentially loaded instead.
162 When the implicit/default collection is generated, it will be named
163 after the module's ``__name__`` attribute, or its last dotted section
164 if it's a submodule. (I.e. it should usually map to the actual ``.py``
165 filename.)
167 Explicitly given collections will only be given that module-derived
168 name if they don't already have a valid ``.name`` attribute.
170 If the module has a docstring (``__doc__``) it is copied onto the
171 resulting `.Collection` (and used for display in help, list etc
172 output.)
174 :param str name:
175 A string, which if given will override any automatically derived
176 collection name (or name set on the module's root namespace, if it
177 has one.)
179 :param dict config:
180 Used to set config options on the newly created `.Collection`
181 before returning it (saving you a call to `.configure`.)
183 If the imported module had a root namespace object, ``config`` is
184 merged on top of it (i.e. overriding any conflicts.)
186 :param str loaded_from:
187 Identical to the same-named kwarg from the regular class
188 constructor - should be the path where the module was
189 found.
191 :param bool auto_dash_names:
192 Identical to the same-named kwarg from the regular class
193 constructor - determines whether emitted names are auto-dashed.
195 .. versionadded:: 1.0
196 """
197 module_name = module.__name__.split(".")[-1]
199 def instantiate(obj_name: Optional[str] = None) -> "Collection":
200 # Explicitly given name wins over root ns name (if applicable),
201 # which wins over actual module name.
202 args = [name or obj_name or module_name]
203 kwargs = dict(
204 loaded_from=loaded_from, auto_dash_names=auto_dash_names
205 )
206 instance = cls(*args, **kwargs)
207 instance.__doc__ = module.__doc__
208 return instance
210 # See if the module provides a default NS to use in lieu of creating
211 # our own collection.
212 for candidate in ("ns", "namespace"):
213 obj = getattr(module, candidate, None)
214 if obj and isinstance(obj, Collection):
215 # TODO: make this into Collection.clone() or similar?
216 ret = instantiate(obj_name=obj.name)
217 ret.tasks = ret._transform_lexicon(obj.tasks)
218 ret.collections = ret._transform_lexicon(obj.collections)
219 ret.default = (
220 ret.transform(obj.default) if obj.default else None
221 )
222 # Explicitly given config wins over root ns config
223 obj_config = copy_dict(obj._configuration)
224 if config:
225 merge_dicts(obj_config, config)
226 ret._configuration = obj_config
227 return ret
228 # Failing that, make our own collection from the module's tasks.
229 tasks = filter(lambda x: isinstance(x, Task), vars(module).values())
230 # Again, explicit name wins over implicit one from module path
231 collection = instantiate()
232 for task in tasks:
233 collection.add_task(task)
234 if config:
235 collection.configure(config)
236 return collection
238 def add_task(
239 self,
240 task: "Task",
241 name: Optional[str] = None,
242 aliases: Optional[Tuple[str, ...]] = None,
243 default: Optional[bool] = None,
244 ) -> None:
245 """
246 Add `.Task` ``task`` to this collection.
248 :param task: The `.Task` object to add to this collection.
250 :param name:
251 Optional string name to bind to (overrides the task's own
252 self-defined ``name`` attribute and/or any Python identifier (i.e.
253 ``.func_name``.)
255 :param aliases:
256 Optional iterable of additional names to bind the task as, on top
257 of the primary name. These will be used in addition to any aliases
258 the task itself declares internally.
260 :param default: Whether this task should be the collection default.
262 .. versionadded:: 1.0
263 """
264 if name is None:
265 if task.name:
266 name = task.name
267 # XXX https://github.com/python/mypy/issues/1424
268 elif hasattr(task.body, "func_name"):
269 name = task.body.func_name # type: ignore
270 elif hasattr(task.body, "__name__"):
271 name = task.__name__
272 else:
273 raise ValueError("Could not obtain a name for this task!")
274 name = self.transform(name)
275 if name in self.collections:
276 err = "Name conflict: this collection has a sub-collection named {!r} already" # noqa
277 raise ValueError(err.format(name))
278 self.tasks[name] = task
279 for alias in list(task.aliases) + list(aliases or []):
280 self.tasks.alias(self.transform(alias), to=name)
281 if default is True or (default is None and task.is_default):
282 self._check_default_collision(name)
283 self.default = name
285 def add_collection(
286 self,
287 coll: "Collection",
288 name: Optional[str] = None,
289 default: Optional[bool] = None,
290 ) -> None:
291 """
292 Add `.Collection` ``coll`` as a sub-collection of this one.
294 :param coll: The `.Collection` to add.
296 :param str name:
297 The name to attach the collection as. Defaults to the collection's
298 own internal name.
300 :param default:
301 Whether this sub-collection('s default task-or-collection) should
302 be the default invocation of the parent collection.
304 .. versionadded:: 1.0
305 .. versionchanged:: 1.5
306 Added the ``default`` parameter.
307 """
308 # Handle module-as-collection
309 if isinstance(coll, ModuleType):
310 coll = Collection.from_module(coll)
311 # Ensure we have a name, or die trying
312 name = name or coll.name
313 if not name:
314 raise ValueError("Non-root collections must have a name!")
315 name = self.transform(name)
316 # Test for conflict
317 if name in self.tasks:
318 err = "Name conflict: this collection has a task named {!r} already" # noqa
319 raise ValueError(err.format(name))
320 # Insert
321 self.collections[name] = coll
322 if default:
323 self._check_default_collision(name)
324 self.default = name
326 def _check_default_collision(self, name: str) -> None:
327 if self.default:
328 msg = "'{}' cannot be the default because '{}' already is!"
329 raise ValueError(msg.format(name, self.default))
331 def _split_path(self, path: str) -> Tuple[str, str]:
332 """
333 Obtain first collection + remainder, of a task path.
335 E.g. for ``"subcollection.taskname"``, return ``("subcollection",
336 "taskname")``; for ``"subcollection.nested.taskname"`` return
337 ``("subcollection", "nested.taskname")``, etc.
339 An empty path becomes simply ``('', '')``.
340 """
341 parts = path.split(".")
342 coll = parts.pop(0)
343 rest = ".".join(parts)
344 return coll, rest
346 def subcollection_from_path(self, path: str) -> "Collection":
347 """
348 Given a ``path`` to a subcollection, return that subcollection.
350 .. versionadded:: 1.0
351 """
352 parts = path.split(".")
353 collection = self
354 while parts:
355 collection = collection.collections[parts.pop(0)]
356 return collection
358 def __getitem__(self, name: Optional[str] = None) -> Any:
359 """
360 Returns task named ``name``. Honors aliases and subcollections.
362 If this collection has a default task, it is returned when ``name`` is
363 empty or ``None``. If empty input is given and no task has been
364 selected as the default, ValueError will be raised.
366 Tasks within subcollections should be given in dotted form, e.g.
367 'foo.bar'. Subcollection default tasks will be returned on the
368 subcollection's name.
370 .. versionadded:: 1.0
371 """
372 return self.task_with_config(name)[0]
374 def _task_with_merged_config(
375 self, coll: str, rest: str, ours: Dict[str, Any]
376 ) -> Tuple[str, Dict[str, Any]]:
377 task, config = self.collections[coll].task_with_config(rest)
378 return task, dict(config, **ours)
380 def task_with_config(
381 self, name: Optional[str]
382 ) -> Tuple[str, Dict[str, Any]]:
383 """
384 Return task named ``name`` plus its configuration dict.
386 E.g. in a deeply nested tree, this method returns the `.Task`, and a
387 configuration dict created by merging that of this `.Collection` and
388 any nested `Collections <.Collection>`, up through the one actually
389 holding the `.Task`.
391 See `~.Collection.__getitem__` for semantics of the ``name`` argument.
393 :returns: Two-tuple of (`.Task`, `dict`).
395 .. versionadded:: 1.0
396 """
397 # Our top level configuration
398 ours = self.configuration()
399 # Default task for this collection itself
400 if not name:
401 if not self.default:
402 raise ValueError("This collection has no default task.")
403 return self[self.default], ours
404 # Normalize name to the format we're expecting
405 name = self.transform(name)
406 # Non-default tasks within subcollections -> recurse (sorta)
407 if "." in name:
408 coll, rest = self._split_path(name)
409 return self._task_with_merged_config(coll, rest, ours)
410 # Default task for subcollections (via empty-name lookup)
411 if name in self.collections:
412 return self._task_with_merged_config(name, "", ours)
413 # Regular task lookup
414 return self.tasks[name], ours
416 def __contains__(self, name: str) -> bool:
417 try:
418 self[name]
419 return True
420 except KeyError:
421 return False
423 def to_contexts(
424 self, ignore_unknown_help: Optional[bool] = None
425 ) -> List[ParserContext]:
426 """
427 Returns all contained tasks and subtasks as a list of parser contexts.
429 :param bool ignore_unknown_help:
430 Passed on to each task's ``get_arguments()`` method. See the config
431 option by the same name for details.
433 .. versionadded:: 1.0
434 .. versionchanged:: 1.7
435 Added the ``ignore_unknown_help`` kwarg.
436 """
437 result = []
438 for primary, aliases in self.task_names.items():
439 task = self[primary]
440 result.append(
441 ParserContext(
442 name=primary,
443 aliases=aliases,
444 args=task.get_arguments(
445 ignore_unknown_help=ignore_unknown_help
446 ),
447 )
448 )
449 return result
451 def subtask_name(self, collection_name: str, task_name: str) -> str:
452 return ".".join(
453 [self.transform(collection_name), self.transform(task_name)]
454 )
456 def transform(self, name: str) -> str:
457 """
458 Transform ``name`` with the configured auto-dashes behavior.
460 If the collection's ``auto_dash_names`` attribute is ``True``
461 (default), all non leading/trailing underscores are turned into dashes.
462 (Leading/trailing underscores tend to get stripped elsewhere in the
463 stack.)
465 If it is ``False``, the inverse is applied - all dashes are turned into
466 underscores.
468 .. versionadded:: 1.0
469 """
470 # Short-circuit on anything non-applicable, e.g. empty strings, bools,
471 # None, etc.
472 if not name:
473 return name
474 from_, to = "_", "-"
475 if not self.auto_dash_names:
476 from_, to = "-", "_"
477 replaced = []
478 end = len(name) - 1
479 for i, char in enumerate(name):
480 # Don't replace leading or trailing underscores (+ taking dotted
481 # names into account)
482 # TODO: not 100% convinced of this / it may be exposing a
483 # discrepancy between this level & higher levels which tend to
484 # strip out leading/trailing underscores entirely.
485 if (
486 i not in (0, end)
487 and char == from_
488 and name[i - 1] != "."
489 and name[i + 1] != "."
490 ):
491 char = to
492 replaced.append(char)
493 return "".join(replaced)
495 def _transform_lexicon(self, old: Lexicon) -> Lexicon:
496 """
497 Take a Lexicon and apply `.transform` to its keys and aliases.
499 :returns: A new Lexicon.
500 """
501 new = Lexicon()
502 # Lexicons exhibit only their real keys in most places, so this will
503 # only grab those, not aliases.
504 for key, value in old.items():
505 # Deepcopy the value so we're not just copying a reference
506 new[self.transform(key)] = copy.deepcopy(value)
507 # Also copy all aliases, which are string-to-string key mappings
508 for key, value in old.aliases.items():
509 new.alias(from_=self.transform(key), to=self.transform(value))
510 return new
512 @property
513 def task_names(self) -> Dict[str, List[str]]:
514 """
515 Return all task identifiers for this collection as a one-level dict.
517 Specifically, a dict with the primary/"real" task names as the key, and
518 any aliases as a list value.
520 It basically collapses the namespace tree into a single
521 easily-scannable collection of invocation strings, and is thus suitable
522 for things like flat-style task listings or transformation into parser
523 contexts.
525 .. versionadded:: 1.0
526 """
527 ret = {}
528 # Our own tasks get no prefix, just go in as-is: {name: [aliases]}
529 for name, task in self.tasks.items():
530 ret[name] = list(map(self.transform, task.aliases))
531 # Subcollection tasks get both name + aliases prefixed
532 for coll_name, coll in self.collections.items():
533 for task_name, aliases in coll.task_names.items():
534 aliases = list(
535 map(lambda x: self.subtask_name(coll_name, x), aliases)
536 )
537 # Tack on collection name to alias list if this task is the
538 # collection's default.
539 if coll.default == task_name:
540 aliases += (coll_name,)
541 ret[self.subtask_name(coll_name, task_name)] = aliases
542 return ret
544 def configuration(self, taskpath: Optional[str] = None) -> Dict[str, Any]:
545 """
546 Obtain merged configuration values from collection & children.
548 :param taskpath:
549 (Optional) Task name/path, identical to that used for
550 `~.Collection.__getitem__` (e.g. may be dotted for nested tasks,
551 etc.) Used to decide which path to follow in the collection tree
552 when merging config values.
554 :returns: A `dict` containing configuration values.
556 .. versionadded:: 1.0
557 """
558 if taskpath is None:
559 return copy_dict(self._configuration)
560 return self.task_with_config(taskpath)[1]
562 def configure(self, options: Dict[str, Any]) -> None:
563 """
564 (Recursively) merge ``options`` into the current `.configuration`.
566 Options configured this way will be available to all tasks. It is
567 recommended to use unique keys to avoid potential clashes with other
568 config options
570 For example, if you were configuring a Sphinx docs build target
571 directory, it's better to use a key like ``'sphinx.target'`` than
572 simply ``'target'``.
574 :param options: An object implementing the dictionary protocol.
575 :returns: ``None``.
577 .. versionadded:: 1.0
578 """
579 merge_dicts(self._configuration, options)
581 def serialized(self) -> Dict[str, Any]:
582 """
583 Return an appropriate-for-serialization version of this object.
585 See the documentation for `.Program` and its ``json`` task listing
586 format; this method is the driver for that functionality.
588 .. versionadded:: 1.0
589 """
590 return {
591 "name": self.name,
592 "help": helpline(self),
593 "default": self.default,
594 "tasks": [
595 {
596 "name": self.transform(x.name),
597 "help": helpline(x),
598 "aliases": [self.transform(y) for y in x.aliases],
599 }
600 for x in sorted(self.tasks.values(), key=lambda x: x.name)
601 ],
602 "collections": [
603 x.serialized()
604 for x in sorted(
605 self.collections.values(), key=lambda x: x.name or ""
606 )
607 ],
608 }