Tasks

class nutcli.tasks.Task(name=None, ignore_errors=False, always=False, enabled=True, timeout=None, logger=None)

Bases: object

Execute operation as a single task.

It can create a nicely formatted output of series of tasks in combination with TaskList.

Note

See TaskList for example usage.

Parameters

  • name (str, optional) – Task name, defaults to None

  • ignore_errors (bool, optional) – If True, all errors will be ignored, defaults to False

  • always (bool, optional) – If True, it will be run inside a TaskList even if some previous tasks raised an exception, defaults to False

  • enabled (bool, optional) – If False, this task will not execute its handler, defaults to True

  • timeout (int or str, optional) – Timeout in seconds or specific format (see nutcli.decorators.Timeout), defaults to None

  • logger (logger, optional) – Logger, defaults to (= nutcli.message)

debug(msg, *args, **kwargs)

Log a debug message.

Parameters

msg (str) – The message.

info(msg, *args, **kwargs)

Log an information message.

Parameters

msg (str) – The message.

warning(msg, *args, **kwargs)

Log a warning message.

Parameters

msg (str) – The message.

error(msg, *args, **kwargs)

Log an error message.

Parameters

msg (str) – The message.

critical(msg, *args, **kwargs)

Log an critical message.

Parameters

msg (str) – The message.

execute(parent=None, **kwargs)

Execute the task’s handler.

Tasks can be nested. Logger is inherited from the parent if it is not None to produce nicely formatted logs.

Parameters

parent (Task, optional) – Parent Task, defaults to None

__call__(function, *args, **kwargs)

Setup a function that will be executed by execute().

If the handler have task among parameters and this parameter is not set in neither positional nor keyword arguments it will be set to the task itself so the handler can access formatted logger functions.

Parameters

function (callable) – Task’s handler.

Returns

Self.

Return type

Task

classmethod Cleanup(*args, **kwargs)

Create a finalizer.

Creates a task that ignore all errors and is executed even if one of the previous tasks failed.

Returns

New task.

Return type

Task

class nutcli.tasks.TaskList(tag=None, name=None, ignore_errors=False, always=False, enabled=True, timeout=None, logger=None, duration=False)

Bases: nutcli.tasks.Task

A task list.

It can execute a list of task in series and provide nicely formatted log output.

This class has wrappers around TaskList.tasks list methods therefore you can use TaskList instance as a list itself to easily add or remove single tasks.

Example usage: Basic use case
tasklist = TaskList()([
    Task('Task 1')(lambda task: task.info('Task 1')),
    Task('Task 2')(lambda task, arg: task.info(arg), 'Task 2'),
    Task('Task 3')(lambda task, arg: task.info(arg), arg='Task 3'),
])

tasklist.execute()

# ->
# [1/3] Task 1
#   Task 1
# [2/3] Task 2
#   Task 2
# [3/3] Task 3
#   Task 3
Example usage: Nested tasks
tasklist = TaskList('task-list')([
    Task('Task 1')(lambda task: task.info('Task 1')),
    Task('Task 2')(lambda task: task.info('Task 2')),
    TaskList('next-level')([
        Task('Task 3')(lambda task: task.info('Task 3')),
        Task('Task 4')(lambda task: task.info('Task 4')),
    ]),
    TaskList()([
        Task('Task 5')(lambda task: task.info('Tag can be empty')),
        Task('Task 6')(lambda task: task.info('It's up to you')),
    ])
])

tasklist.execute()

# ->
# [task-list] [1/4] Task 1
# [task-list]   Task 1
# [task-list] [2/4] Task 2
# [task-list]   Task 2
# [task-list] [3/4]
# [task-list]   [next-level] [1/2] Task 3
# [task-list]   [next-level]   Task 3
# [task-list]   [next-level] [2/2] Task 4
# [task-list]   [next-level]   Task 4
# [task-list] [4/4]
# [task-list]   [1/2] Task 5
# [task-list]     Tag can be empty
# [task-list]   [2/2] Task 6
# [task-list]     It's up to you
Example usage: Finalizer
def raise_task():
    raise Exception('Ooops')

tasklist = TaskList('task-list')([
    Task('Task 1')(lambda task: task.info('I am doing something bad')),
    Task('Task 2')(raise_task),
    Task('Task 3')(lambda task: task.info('I was skipped')),
    Task.Cleanup('Task 4')(lambda task: task.info('Cleaning up'))
])

tasklist.execute()

# ->
# [task-list] [1/4] Task 1
# [task-list]   I am doing something bad
# [task-list] [2/4] Task 2
# [task-list] ERROR Exception: Ooops
# [task-list] [3/4] Task 3 (skipped on error)
# [task-list] [4/4] Task 4 (finalizing)
# [task-list]   Cleaning up
Parameters

  • tag (str, optional) – Tag that will be visible on each log message, defaults to None

  • name (str, optional) – Task name, defaults to None

  • ignore_errors (bool, optional) – If True, all errors will be ignored, defaults to False

  • always (bool, optional) – If True, it will be run inside a TaskList even if some previous tasks raised an exception, defaults to False

  • enabled (bool, optional) – If False, this task will not execute its handler, defaults to True

  • timeout (int or str, optional) – Timeout in seconds or specific format (see nutcli.decorators.Timeout), defaults to None

  • logger (logger, optional) – Logger, defaults to (= nutcli.message)

  • duration (bool, optional) – If True, the time that the task list takes to finish is printed at the end, defaults to False

__call__(tasks)

Add tasks to the list.

Parameters

tasks – List of tasks.

Type

list of Task

Returns

Self.

Return type

TaskList

append(object, /)

Append object to the end of the list.

extend(iterable, /)

Extend list by appending elements from the iterable.

insert(index, object, /)

Insert object before index.

remove(value, /)

Remove first occurrence of value.

Raises ValueError if the value is not present.

pop(index=-1, /)

Remove and return item at index (default last).

Raises IndexError if list is empty or index is out of range.

clear()

Remove all items from list.