vmklib.tasks
index
/home/vkottler/src/vkottler/workspace/vmklib/vmklib/tasks/__init__.py

A module implementing a base task.

 
Package Contents
       
args
clean
github
mixins (package)
node (package)
python (package)
release
venv

 
Classes
       
vcorelib.task.Task(vcorelib.logging.LoggerMixin)
VmklibBase

 
class VmklibBase(vcorelib.task.Task)
    VmklibBase(name: 'str', *args, execute: 'TaskExecute' = None, log: 'Logger' = None, timer: 'Timer' = None, target: 'Target' = None, **kwargs) -> 'None'
 
Create a concrete file output after a task completes.
 
 
Method resolution order:
VmklibBase
vcorelib.task.Task
vcorelib.logging.LoggerMixin
builtins.object

Class methods defined here:
build_dir(inbox: Dict[str, Dict[str, Any]]) -> pathlib.Path from builtins.type
Get a path to the build directory.

Data and other attributes defined here:
__annotations__ = {}
default_requirements = {'vmklib.init'}

Methods inherited from vcorelib.task.Task:
__init__(self, name: 'str', *args, execute: 'TaskExecute' = None, log: 'Logger' = None, timer: 'Timer' = None, target: 'Target' = None, **kwargs) -> 'None'
Initialize this task.
__str__(self) -> 'str'
Convert this task into a string.
create_execute(self, *args, **kwargs) -> 'TaskExecute'
Create a default execute function for this task.
depend_on(self, task: "'Task'", eloop: '_asyncio.AbstractEventLoop' = None, **kwargs) -> 'bool'
Register other tasks' output data to your input box. Return true
if a new dependency was added.
depend_on_all(self, tasks: "_Iterable['Task']", **kwargs) -> 'int'
Register multiple dependencies. Return the number of dependencies
added.
async dispatch(self, caller: "'Task'" = None, init_only: 'bool' = False, substitutions: 'Substitutions' = None, **kwargs) -> 'None'
Dispatch this task and return whether or not it succeeded.
dispatch_init(self, caller: "'Task'" = None, substitutions: 'Substitutions' = None, **kwargs) -> '_Tuple[Substitutions, str]'
Perform some dispatch initialization.
resolve(self, log: 'Logger', compiled: 'str', substitutions: 'Substitutions' = None) -> 'None'
Mark this task resolved.
async resolve_dependencies(self, **substitutions) -> 'None'
A default dependency resolver for tasks. Note that the default
dependency resolver cannot propagate current-task substitutions to
its dependencies as they've already been explicitly registered.
resolved(self, compiled: 'str', substitutions: 'Substitutions' = None) -> 'bool'
Override this in a derived task to implement more complex logic.
async run(self, inbox: 'Inbox', outbox: 'Outbox', *args, **kwargs) -> 'bool'
Override this method to implement the task.
async run_enter(self, _inbox: 'Inbox', _outbox: 'Outbox', *_args, **_kwargs) -> 'bool'
A default enter method.
async run_exit(self, _inbox: 'Inbox', _outbox: 'Outbox', *_args, **_kwargs) -> 'bool'
A default exit method.
task_fail(self, compiled: 'str', substitutions: 'Substitutions', caller: "'Task'" = None, indent: 'int' = 4) -> 'TaskFailed'
Build an exception message for when this task fails.

Readonly properties inherited from vcorelib.task.Task:
stack
Get this task's execution stack.

Data and other attributes inherited from vcorelib.task.Task:
stages = ['run_enter', 'run', 'run_exit']

Methods inherited from vcorelib.logging.LoggerMixin:
log_time(self, message: str, *args, level: int = 20, reminder: bool = False, **kwargs) -> Iterator[NoneType]
A simple wrapper.

Data descriptors inherited from vcorelib.logging.LoggerMixin:
__dict__
dictionary for instance variables
__weakref__
list of weak references to the object

 
Data
        Inbox = typing.Dict[str, typing.Dict[str, typing.Any]]