Task Module

Task module.

class pepperpy.task.Task(task_id, callback)[source]

Bases: Generic[T]

Task.

Parameters:

task_id (str)
callback (Callable[[], Awaitable[T]])

__init__(task_id, callback)[source]

Initialize task.

Parameters:

task_id (str) – Task ID
callback (Callable[[], Awaitable[TypeVar(T)]]) – Task callback

Return type:

None

property state: TaskState

Get task state.

Returns:: Task state

property result: TaskResult[T] | None

Get task result.

Returns:: Task result

async run()[source]

Run task.

Return type:: TaskResult[TypeVar(T)]
Returns:: Task result
Raises:: TaskError – If task is already running or cancelled

cancel()[source]

Cancel task.

Raises:: TaskError – If task is already completed or failed
Return type:: None

class pepperpy.task.TaskConfig(name='task_manager', metadata=<factory>)[source]

Bases: ModuleConfig

Task configuration.

Parameters:

name (str)
metadata (Dict[str, Any])

name: str = 'task_manager'

metadata: Dict[str, Any]

exception pepperpy.task.TaskError(message, details=None, cause=None)[source]

Bases: PepperpyError

Task error.

Parameters:

message (str)
details (Dict[str, Any] | None)
cause (Exception | None)

Return type:

None

class pepperpy.task.TaskManager(config=None)[source]

Bases: BaseModule[TaskConfig]

Task manager.

Parameters:: config (TaskConfig | None)

__init__(config=None)[source]

Initialize task manager.

Parameters:: config (Optional[TaskConfig]) – Task configuration
Return type:: None

submit(task_id, callback)[source]

Submit task.

Parameters:

task_id (str) – Task ID
callback (Callable[[], Awaitable[TypeVar(T)]]) – Task callback

Return type:

Task[TypeVar(T)]

Returns:

Task instance

Raises:

TaskError – If task cannot be submitted

get(task_id)[source]

Get task.

Parameters:: task_id (str) – Task ID
Return type:: Task[Any]
Returns:: Task instance
Raises:: TaskError – If task is not found

cancel(task_id)[source]

Cancel task.

Parameters:: task_id (str) – Task ID
Raises:: TaskError – If task cannot be cancelled
Return type:: None

class pepperpy.task.TaskResult(task_id, state, result=None, error=None)[source]

Bases: Generic[T]

Task result.

Parameters:

task_id (str)
state (TaskState)
result (T | None)
error (Exception | None)

task_id: str

state: TaskState

result: Optional[TypeVar(T)] = None

error: Optional[Exception] = None

class pepperpy.task.TaskState(value, names=<not given>, *values, module=None, qualname=None, type=None, start=1, boundary=None)[source]

Bases: Enum

Task state.

PENDING = 1

RUNNING = 2

COMPLETED = 3

FAILED = 4

CANCELLED = 5