Task

Decorator to turn a function or coroutine function into a task.

def task(
    f: Union[None, Callable[P, Union[Coroutine[Any, Any, R], R]]] = None,
    *,
    prefer_threaded: bool = True,
) -> Union[Callable[[Callable[P, R]], Task[P, R]], Task[P, R]]:

    ...

Lets you run code in the background, with the UI available to the user. This is useful for long running tasks, like downloading data or processing data.

The task decorator turns a function or coroutine function (async def foo(...) - here foo is called a coroutine function) into a task object. A task is a callable that will run the function or coroutine function in a separate thread Note that on platforms where threads are supported, asyncio tasks will still be executed in threads (unless the prefer_thread=False argument is passed). Because a coroutine function might still call long running blocking code. Running the asyncio task in a thread will still result in a responsive UI when executed in a separate thread.

The task object will execute the function only once per virtual kernel and will only store one result per virtual kernel. When called multiple times, the previously started thread or asyncio task result will be ignored.

A running thread or asyncio task can check if it is still the current task by calling task.is_current(). If task.is_current() returns False, the task should stop running and return early.

The return value of the function is available as the .value reactive property on the task object, meaning that if a component accesses it, the component will automatically re-run when the value changes, like a reactive variable.

Task object

The task object has the following attributes/values which are all reactive:

• .value: Contains the return value of the function (Only valid if .finished is true, else None).
• .exception: The exception raised by the function, if any (Only valid if .error is true, else None).
• .latest The last return value of the function, useful for showing out-of-date data while the task is running.
• .progress A readable and writable reactive property which can be used for communicating progress to the user.

The state of the task can be queried with the following attributes, which are all reactive:

• .not_called: True if the task has not been called yet.
• .pending: True if the task is asked to run, but did not finish yet, did not error and did not get cancelled. When true, often a loading or busy indicator is shown to the user.
• .finished: True if the task has finished running. The result is available in the .value attribute as well as the .latest attribute.
• .cancelled: True if the task was cancelled (by calling .cancel()).
• .error: True if the function has raised an exception.

The following methods are available:

• (*args, **kwargs) : Call the task with the given arguments and keyword arguments. The task will only run once per virtual kernel.
• .cancel(): Cancels the task.
• is_current(): Returns True if the task is still the current task, and should continue running. Will return False when a new call to the task is made, and this function is being called from the the previous thread or asyncio.

State diagram

The following state diagram shows the possible states of a task and how each state transitions to another state.

Note that calling the task (as indicated by task()) can be done from any state.

Example

Async task

import asyncio
import solara
from solara.lab import task

@task
async def fetch_data():
    await asyncio.sleep(2)
    return "The answer is 42"

@solara.component
def Page():
    solara.Button("Fetch data", on_click=fetch_data)
    solara.ProgressLinear(fetch_data.pending)

    if fetch_data.finished:
        solara.Text(fetch_data.value)
    elif fetch_data.not_called:
        solara.Text("Click the button to fetch data")
    # Optional state check
    # elif fetch_data.cancelled:
    #     solara.Text("Cancelled the fetch")
    # elif fetch_data.error:
    #     solara.Error(str(fetch_data.exception))

Threaded task

import time
import solara
from solara.lab import task

@task
def fetch_data():
    time.sleep(2)
    return "The answer is 42"


@solara.component
def Page():
    solara.Button("Fetch data", on_click=fetch_data)
    solara.ProgressLinear(fetch_data.pending)

    if fetch_data.finished:
        solara.Text(fetch_data.value)
    elif fetch_data.not_called:
        solara.Text("Click the button to fetch data")
    # Optional state check
    # elif fetch_data.cancelled:
    #     solara.Text("Cancelled the fetch")
    # elif fetch_data.error:
    #    solara.Error(str(fetch_data.exception))

Note that both examples are very similar. In the first example however, we wrap a coroutine function which can use asyncio.sleep. In the second example, we use a regular function, which uses time.sleep. If the coroutine function would use time.sleep in combination with prefer_threaded=False, the UI would be unresponsive for 2 seconds.

Showing a progress bar

Using the .progress attribute, you can show a progress bar to the user. This is useful for long running tasks but requires a bit more work.

import time
import solara
from solara.lab import task


@task
def my_calculation():
    total = 0
    for i in range(10):
        my_calculation.progress = (i + 1) * 10.0
        time.sleep(0.4)
        if not my_calculation.is_current():
            # a new call was made before this call was finished
            return
        total += i**2
    return total


@solara.component
def Page():
    solara.Button("Run calculation", on_click=my_calculation)
    solara.ProgressLinear(my_calculation.progress if my_calculation.pending else False)

    if my_calculation.finished:
        solara.Text(f"Calculation result: {my_calculation.value}")
    elif my_calculation.not_called:
        solara.Text("Click the button to fetch data")
    # Optional state check
    # elif my_calculation.cancelled:
    #     solara.Text("Cancelled the fetch")
    # elif my_calculation.error:
    #    solara.Error(str(my_calculation.exception))

Out-of-date data

import time
import solara
from solara.lab import task


@task
def my_calculation():
    total = 0
    for i in range(10):
        time.sleep(0.1)
        total += i**2
    return total


@solara.component
def Page():
    solara.ProgressLinear(my_calculation.pending)
    solara.Button("Run simulation", on_click=my_calculation)
    print(my_calculation.pending, my_calculation.value)

    if my_calculation.finished:
        solara.Text(f"Simulation result: {my_calculation.value}")
    if my_calculation.pending and my_calculation.latest:
        solara.Text(f"Simulation previous result: {my_calculation.latest}", style={"opacity": ".3"})
    elif my_calculation.not_called:
        solara.Text("Click the button to fetch data")

Arguments

• f: Function to turn into task or None
• prefer_threaded - bool: Will run coroutine functions as a task in a thread when threads are available. This ensures that even when a coroutine functions calls a blocking function the UI is still responsive. On platform where threads are not supported (like Pyodide / WASM / Emscripten / PyScript), a coroutine function will always run in the current event loop.

```