Tasks

class nutcli.tasks.Task(name=None, ignore_errors=False, always=False, taskarg=True, 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

  • taskarg (bool, optional) – If True, the first argument of task’s handler will be the task itself, defaults to True

  • 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().

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, timeout=None, logger=None)

Bases: nutcli.tasks.Task

A task list.

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

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', taskarg=False)(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

  • 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)

__call__(tasks)

Add tasks to the list.

Parameters

tasks – List of tasks.

Type

list of Task

Returns

Self.

Return type

TaskList