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.