solara-threading

Decision Guide

| API | Scope | Status | Use when | |---|---|---|---| | use_task (from solara.lab) | Component-local | Recommended | Background work in a component | | @task (from solara.lab) | Global/shared | Stable | App-wide background jobs shared across components | | use_thread | Component-local | Deprecated | Legacy code only — migrate to use_task |

use_task (Recommended)

Runs a sync function in a thread or an async coroutine as a task. Returns a Task[T] object.

import time
import solara
from solara.lab import use_task

@solara.component
def SquareCalculator():
    number = solara.use_reactive(4)

    def compute():
        time.sleep(1)  # simulate expensive work
        return number.value ** 2

    result = use_task(compute, dependencies=[number.value])

    solara.InputInt("Number", value=number)
    solara.ProgressLinear(result.pending)
    if result.finished:
        solara.Success(f"Result: {result.value}")
    if result.error:
        solara.Error(f"Error: {result.exception}")

Async variant — the only difference is async def + await:

import asyncio
from solara.lab import use_task

@solara.component
def AsyncExample():
    async def fetch():
        await asyncio.sleep(1)
        return {"status": "ok"}

    result = use_task(fetch, dependencies=[])

Key parameters

• dependencies — list of values; task re-runs when any change. None = manual invocation only.
• prefer_threaded (default True) — runs coroutines in a thread so blocking calls don't freeze the UI.
• raise_error (default True) — if False, errors are captured in result.exception instead of raised.

Task object attributes

| Attribute | Type | Description | |---|---|---| | .value | T | Result (available when finished) | | .pending | bool | True while running | | .finished | bool | True when complete | | .error | bool | True if exception occurred | | .exception | Exception | The exception (if any) | | .cancel() | method | Cancel the running task | | .retry() | method | Re-run the task |

@task Decorator

Like use_task but the result is globally shared — all components see the same state:

from solara.lab import task

@task
def train_model(epochs: int):
    for i in range(epochs):
        time.sleep(1)  # training step
    return {"accuracy": 0.95}

@solara.component
def TrainButton():
    solara.Button("Train", on_click=lambda: train_model(10))
    solara.ProgressLinear(train_model.pending)

@solara.component
def ResultPanel():
    if train_model.finished:
        solara.Text(f"Accuracy: {train_model.value['accuracy']}")

use_thread (Deprecated)

Returns Result[T] with states: RUNNING, FINISHED, ERROR, CANCELLED, WAITING, STARTING.

from solara import use_thread

@solara.component
def LegacyExample():
    def work(cancel: threading.Event):
        while not cancel.is_set():
            # do work, check cancel periodically
            pass
        return result

    result = use_thread(work, dependencies=[])

intrusive_cancel=True (default) installs a tracer to force-cancel threads — significant performance overhead. Prefer use_task which avoids this.

Anti-Patterns

| Avoid | Do Instead | |---|---| | Raw threading.Thread | Use use_task — preserves Solara context | | time.sleep() in render body | Move to use_task or use_effect | | Accessing result.value without checking .pending | Always guard with if result.finished: | | use_thread in new code | Migrate to use_task | | Blocking coroutines without prefer_threaded | Keep prefer_threaded=True (default) |

When to Use

Load this skill when implementing background tasks, async operations, or any work that should not block the Solara UI.