Skip to content

Worker

A class to manage concurrent work.

WorkType module-attribute

WorkType: TypeAlias = Union[
    Callable[[], Coroutine[None, None, ResultType]],
    Callable[[], ResultType],
    Awaitable[ResultType],
]

Type used for workers.

active_worker module-attribute

active_worker: ContextVar[Worker] = ContextVar(
    "active_worker"
)

Currently active worker context var.

DeadlockError class

Bases: WorkerError

The operation would result in a deadlock.

NoActiveWorker class

Bases: Exception

There is no active worker.

Worker class

def __init__(
    self,
    node,
    work=None,
    *,
    name="",
    group="default",
    description="",
    exit_on_error=True
):

Bases: Generic[ResultType]

A class to manage concurrent work (either a task or a thread).

Parameters
Name Type Description Default
node DOMNode

The widget, screen, or App that initiated the work.

required
work WorkType | None

A callable, coroutine, or other awaitable object to run in the worker.

None
name str

Name of the worker (short string to help identify when debugging).

''
group str

The worker group.

'default'
description str

Description of the worker (longer string with more details).

''
exit_on_error bool

Exit the app if the worker raises an error. Set to False to suppress exceptions.

True

completed_steps property

completed_steps: int

The number of completed steps.

is_cancelled property

is_cancelled: bool

Has the work been cancelled?

Note that cancelled work may still be running.

is_finished property

is_finished: bool

Has the task finished (cancelled, error, or success)?

is_running property

is_running: bool

Is the task running?

node property

node: DOMNode

The node where this worker was run from.

progress property

progress: float

Progress as a percentage.

If the total steps is None, then this will return 0. The percentage will be clamped between 0 and 100.

result property

result: ResultType | None

The result of the worker, or None if there is no result.

state writable property

state: WorkerState

The current state of the worker.

total_steps property

total_steps: int | None

The number of total steps, or None if indeterminate.

StateChanged class

def __init__(self, worker, state):

Bases: Message

The worker state changed.

Parameters
Name Type Description Default
worker Worker

The worker object.

required
state WorkerState

New state.

required

advance method

def advance(self, steps=1):

Advance the number of completed steps.

Parameters
Name Type Description Default
steps int

Number of steps to advance.

1

cancel method

def cancel(self):

Cancel the task.

run async

def run(self):

Run the work.

Implement this method in a subclass, or pass a callable to the constructor.

Returns
Type Description
ResultType

Return value of work.

update method

def update(self, completed_steps=None, total_steps=-1):

Update the number of completed steps.

Parameters
Name Type Description Default
completed_steps int | None

The number of completed seps, or None to not change.

None
total_steps int | None

The total number of steps, None for indeterminate, or -1 to leave unchanged.

-1

wait async

def wait(self):

Wait for the work to complete.

Raises
Type Description
WorkerFailed

If the Worker raised an exception.

WorkerCancelled

If the Worker was cancelled before it completed.

Returns
Type Description
ResultType

The return value of the work.

WorkerCancelled class

Bases: WorkerError

The worker was cancelled and did not complete.

WorkerError class

Bases: Exception

A worker related error.

WorkerFailed class

def __init__(self, error):

Bases: WorkerError

The worker raised an exception and did not complete.

WorkerState class

Bases: enum.Enum

A description of the worker's current state.

CANCELLED class-attribute instance-attribute

CANCELLED = 3

Worker is not running, and was cancelled.

ERROR class-attribute instance-attribute

ERROR = 4

Worker is not running, and exited with an error.

PENDING class-attribute instance-attribute

PENDING = 1

Worker is initialized, but not running.

RUNNING class-attribute instance-attribute

RUNNING = 2

Worker is running.

SUCCESS class-attribute instance-attribute

SUCCESS = 5

Worker is not running, and completed successfully.

get_current_worker function

def get_current_worker():

Get the currently active worker.

Raises
Type Description
NoActiveWorker

If there is no active worker.

Returns
Type Description
Worker

A Worker instance.