Running Background Work in Textual with Workers

A terminal UI that freezes while it loads data feels broken: keys do nothing, the screen does not redraw, Ctrl-C seems ignored. In Textual, anything slow — an HTTP call, a database query, parsing a large file — must run outside the event loop, in a worker. Textual's @work decorator makes that easy, but the defaults and the details decide whether the result is robust: by default a worker that raises an exception takes the whole app down; results from a cancelled worker can still arrive after a newer one has started; widgets must only be touched from the main thread. This guide builds a job browser that loads pages in a thread worker with a progress bar, reloads on demand without mixing old and new results, reports failures in the UI instead of crashing, and tests all of it with Pilot. It belongs to the Textual topic.

Prerequisites

• Textual 1.0+ (checked with 8.2); pytest with pytest-asyncio in asyncio_mode = "auto".
• The app structure from building your first Textual app.

Thread workers and async workers

@work turns a method into a worker launcher: calling it starts the work in the background and returns a Worker object immediately. There are two kinds. An async worker (@work on an async def) runs on the event loop and suits async libraries such as httpx.AsyncClient; it must still await regularly, or it blocks the UI like any other coroutine. A thread worker (@work(thread=True) on a regular function) runs in a separate thread and suits blocking code — requests, sqlite3, file parsing, most SDKs. The rule from Textual's documentation applies to both: only the main thread may update widgets, so thread workers hand updates back with self.call_from_thread(...).

The recipe

# src/mytool/jobs_app.py
from __future__ import annotations

import time
from collections.abc import Callable

from textual import work
from textual.app import App, ComposeResult
from textual.widgets import DataTable, Footer, Label, ProgressBar
from textual.worker import Worker, WorkerState, get_current_worker


def slow_fetch(page: int) -> list[dict]:
    """Stand-in for a blocking API call."""
    time.sleep(0.01)
    return [{"id": page * 10 + i, "status": "ok"} for i in range(10)]


class Jobs(App[None]):
    BINDINGS = [("r", "reload", "Reload")]

    def __init__(self, fetch: Callable[[int], list[dict]] = slow_fetch, pages: int = 5) -> None:
        super().__init__()
        self.fetch = fetch
        self.pages = pages

    def compose(self) -> ComposeResult:
        yield Label("loading…", id="status")
        yield ProgressBar(total=self.pages, id="progress")
        yield DataTable(id="table")
        yield Footer()

    def on_mount(self) -> None:
        self.query_one(DataTable).add_columns("id", "status")
        self.load()

    @work(thread=True, exclusive=True, group="load", exit_on_error=False)
    def load(self) -> int:
        worker = get_current_worker()
        self.call_from_thread(self.query_one(DataTable).clear)
        total = 0
        for page in range(self.pages):
            if worker.is_cancelled:
                break
            rows = self.fetch(page)
            total += len(rows)
            self.call_from_thread(self.add_rows, worker, rows, page + 1)
        return total

    def add_rows(self, worker: Worker, rows: list[dict], done: int) -> None:
        if worker.is_cancelled:            # a newer load replaced this one; drop late results
            return
        self.query_one(DataTable).add_rows([(r["id"], r["status"]) for r in rows])
        self.query_one(ProgressBar).update(progress=done)

    def on_worker_state_changed(self, event: Worker.StateChanged) -> None:
        status = self.query_one("#status", Label)
        if event.state is WorkerState.SUCCESS:
            status.update(f"{event.worker.result} jobs")
        elif event.state is WorkerState.ERROR:
            status.update(f"[red]failed:[/red] {event.worker.error}")

    def action_reload(self) -> None:
        self.load()

Each argument to @work matters:

• thread=True because fetch blocks. In an async app with an async client, drop it and make load a coroutine.
• exclusive=True with a group cancels any running worker in the same group when a new one starts, so pressing r repeatedly never runs several loads at once.
• exit_on_error=False is the important one. The default is True: an exception in the worker exits the application and prints a traceback. For a load that can fail because a network is down, that is almost never what you want. With False, the worker moves to the ERROR state and the app decides what to show.

Cancellation is cooperative — and results can arrive late

exclusive=True requests cancellation of the old worker; a thread cannot be stopped from outside, so the worker must check worker.is_cancelled and stop on its own. Between the cancel request and that check, the old worker can still finish a page and schedule add_rows — after the new worker has already cleared the table. The first version of this app showed exactly that: an occasional extra page of rows after a reload, visible only under timing pressure in a test. Passing the worker into add_rows and dropping results from cancelled workers on the main thread closes the race for good.

Progress and state

Progress updates follow the same route as data: from the thread through call_from_thread, applied on the main thread. The worker's outcome arrives as a Worker.StateChanged message, handled in on_worker_state_changed: SUCCESS carries the return value in worker.result, ERROR carries the exception in worker.error, and CANCELLED lets you reset a spinner. Handling outcomes in one place keeps the worker function free of UI code beyond the progress hand-offs.

The async variant

With an async HTTP client, the same load is an async worker, and two things get simpler:

@work(exclusive=True, group="load", exit_on_error=False)
async def load_async(self) -> int:
    async with httpx.AsyncClient(base_url="https://api.example.com", timeout=10) as client:
        response = await client.get("/jobs")
        response.raise_for_status()
        jobs = response.json()
    table = self.query_one(DataTable)
    table.clear()
    table.add_rows((job["id"], job["status"]) for job in jobs)
    return len(jobs)

Async workers run on the event loop, so they may update widgets directly — no call_from_thread. And cancellation is real rather than cooperative: an exclusive reload raises CancelledError inside the old worker at its next await, so it never reaches the lines that touch the table, and the late-results race cannot happen. The price is that every slow step must be awaitable; one blocking call inside an async worker freezes the whole interface.

UX considerations

• Show that something is happening. A progress bar when the amount of work is known, a LoadingIndicator or "loading…" label when it is not; never a silent, empty screen.
• Keep the UI usable during loads. Scrolling, filtering already-loaded rows and quitting should all work while the worker runs — the whole point of moving work off the event loop.
• Report failures where the data would be. "failed: API unreachable — press r to retry" in the status line is better than a modal, and much better than a crash.
• Cancel on exit. Textual cancels workers when the app exits, but long blocking calls inside a thread still run to completion; give blocking calls timeouts so quitting is prompt.

Testing the behaviour

app.workers.wait_for_complete() waits for background work without sleeping. Note that it re-raises a worker's exception as WorkerFailed even when the app itself handled it, so tests of error paths suppress that:

# tests/test_jobs_app.py
import contextlib
import time

from textual.worker import WorkerFailed

from mytool.jobs_app import Jobs


async def test_loads_all_pages_in_background():
    app = Jobs(pages=3)
    async with app.run_test() as pilot:
        await app.workers.wait_for_complete()
        await pilot.pause()
        assert app.query_one("DataTable").row_count == 30
        assert "30 jobs" in str(app.query_one("#status").render())


async def test_error_is_shown_not_raised():
    def broken(page):
        raise ConnectionError("API unreachable")

    app = Jobs(fetch=broken)
    async with app.run_test() as pilot:
        with contextlib.suppress(WorkerFailed):      # the waiter re-raises; the app does not
            await app.workers.wait_for_complete()
        await pilot.pause()
        assert "API unreachable" in str(app.query_one("#status").render())


async def test_reload_replaces_rather_than_appends():
    def tracking(page):
        time.sleep(0.02)
        return [{"id": page, "status": "ok"}]

    app = Jobs(fetch=tracking, pages=20)
    async with app.run_test() as pilot:
        await pilot.press("r")                      # reload while the first load runs
        await app.workers.wait_for_complete()
        await pilot.pause()
        assert app.query_one("DataTable").row_count == 20

The reload test is the one that caught the late-results race: it failed intermittently, in roughly one run in three, until add_rows learned to ignore cancelled workers. Run tests like it several times in a row when you change worker code; timing bugs rarely show up on the first attempt.

Conclusion

Workers keep a Textual app responsive: use thread workers for blocking code and async workers for async libraries, update widgets only on the main thread via call_from_thread, make reloads exclusive within a group, set exit_on_error=False and show failures through Worker.StateChanged, and drop results from cancelled workers so a reload never mixes old and new data. Test with wait_for_complete(), remembering that it re-raises worker errors, and repeat timing-sensitive tests.

Frequently asked questions

Related

• Up: Building terminal UIs with Textual
• Browsing records with a Textual DataTable
• Testing Textual apps with Pilot
• Parallelising CLI work with thread pools
• Cancelling async tasks on Ctrl-C