Subpackages#

Submodules#

vcorelib.task.manager module#

A simple task management interface.

class vcorelib.task.manager.TaskManager(resolver: TargetResolver = None)[source]#

Bases: ScriptableMixin

A class for managing concurrent execution of tasks and also interfacing them via names.

evaluate(tasks: Iterable[str], **kwargs) → Tuple[List[Tuple[Task, TargetMatch]], Set[str]][source]#

Iterate over task strings and provide a list of tasks that can be executed (plus relevant data from matching the target) as well as the set of tasks that can’t be resolved.

execute(tasks: Iterable[str], **kwargs) → Set[str][source]#

Execute some set of provided tasks. Return the tasks that don’t get resolved.

finalize(**kwargs) → None[source]#

Register task dependencies while the event loop is running.

prepare_execute(tasks: Iterable[str], **kwargs) → Tuple[Set[str], Callable[[], Coroutine[Any, Any, None]]][source]#

Gather tasks that can and can’t be executed to prepare a function that can execute tasks and also return the task strings that wouldn’t be resolved.

register(task: Task, dependencies: Iterable[str] = None, target: str = None) → bool[source]#

Register a new task and apply any requested dependencies.

register_to(target: str, dependencies: Iterable[str]) → bool[source]#

Register dependencies to a task by name.

Module contents#

A module for implementing tasks in a dependency tree.

class vcorelib.task.FailTask(name: str, *args, execute: Callable[[Dict[str, Dict[str, Any]], Dict[str, Any]], Coroutine[Any, Any, bool]] = None, log: Logger = None, timer: Timer = None, target: Target = None, **kwargs)[source]#

Bases: Task

A task that always fails.

async run(inbox: Dict[str, Dict[str, Any]], outbox: Dict[str, Any], *args, **kwargs) → bool[source]#

Task fails by default.

class vcorelib.task.Phony(name: str, *args, execute: Callable[[Dict[str, Dict[str, Any]], Dict[str, Any]], Coroutine[Any, Any, bool]] = None, log: Logger = None, timer: Timer = None, target: Target = None, **kwargs)[source]#

Bases: Task

A task stub that doesn’t do anything. Useful for top-level targets that depend on other concrete tasks.

async run(inbox: Dict[str, Dict[str, Any]], outbox: Dict[str, Any], *args, **kwargs) → bool[source]#

Override this method to implement the task.

class vcorelib.task.Task(name: str, *args, execute: Callable[[Dict[str, Dict[str, Any]], Dict[str, Any]], Coroutine[Any, Any, bool]] = None, log: Logger = None, timer: Timer = None, target: Target = None, **kwargs)[source]#

Bases: LoggerMixin

A basic task interface definition.

create_execute(*args, **kwargs) → Callable[[Dict[str, Dict[str, Any]], Dict[str, Any]], Coroutine[Any, Any, bool]][source]#

Create a default execute function for this task.

default_requirements: _Set[str] = {}#

depend_on(task: Task, eloop: AbstractEventLoop = None, **kwargs) → bool[source]#

Register other tasks’ output data to your input box. Return true if a new dependency was added.

depend_on_all(tasks: Iterable[Task], **kwargs) → int[source]#

Register multiple dependencies. Return the number of dependencies added.

async dispatch(caller: Task = None, init_only: bool = False, substitutions: Dict[str, str | int] = None, **kwargs) → None[source]#

Dispatch this task and return whether or not it succeeded.

dispatch_init(caller: Task = None, substitutions: Dict[str, str | int] = None, **kwargs) → Tuple[Dict[str, str | int], str][source]#

Perform some dispatch initialization.

resolve(log: Logger, compiled: str, substitutions: Dict[str, str | int] = None) → None[source]#

Mark this task resolved.

async resolve_dependencies(**substitutions) → None[source]#

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(compiled: str, substitutions: Dict[str, str | int] = None) → bool[source]#

Override this in a derived task to implement more complex logic.

async run(inbox: Dict[str, Dict[str, Any]], outbox: Dict[str, Any], *args, **kwargs) → bool[source]#

Override this method to implement the task.

async run_enter(_inbox: Dict[str, Dict[str, Any]], _outbox: Dict[str, Any], *_args, **_kwargs) → bool[source]#

A default enter method.

async run_exit(_inbox: Dict[str, Dict[str, Any]], _outbox: Dict[str, Any], *_args, **_kwargs) → bool[source]#

A default exit method.

property stack: ExitStack#

Get this task’s execution stack.

stages = ['run_enter', 'run', 'run_exit']#

task_fail(compiled: str, substitutions: Dict[str, str | int], caller: Task = None, indent: int = 4) → TaskFailed[source]#

Build an exception message for when this task fails.

exception vcorelib.task.TaskFailed[source]#

Bases: Exception

A custom exception to indicate that a task failed.