Background tasks

A UI must never block waiting on the network or the disk. Pyzed offers two tools:

• Command.perform(fn, then) runs a blocking function on another thread and hands you the result.
• pz.post(message) sends messages to the app from any thread (progress, events).

Goal	Time	Level	You need
Downloads with progress bars and notifications that never freeze the window	30 minutes	Intermediate · Python	Dialogs and notifications

What you’ll learn

• Why slow work must leave the UI thread, and the two ways to move it out.
• Get one result back with Command.perform.
• Report progress from a thread with pz.post and handle it in update.
• Show notifications that dismiss themselves.

Tip

If you prefer asyncio, update and callbacks can also be async def functions or async generators, and pz.listen turns a stream into a subscription. See the Async and subscriptions guide.

1. One result: Command.perform. For “do this and tell me when it’s done”:

   def check_updates(self):
       self.checking = True
       return Command.perform(self.fetch_latest_version, self.got_version)
   
   @staticmethod
   def fetch_latest_version():
       time.sleep(1.5)          # e.g. an HTTP request
       return "0.2.0"
   
   def got_version(self, version):     # runs on the UI thread
       self.checking, self.latest = False, version

   While self.checking is True, the button is drawn disabled (on_press=None).

2. Many results: pz.post. A thread of your own reports progress with messages that reach update:

   def start(self, name):
       self.progress[name] = 0.0
   
       def worker():
           done = 0.0
           while done < 1.0:
               time.sleep(0.05)
               done = min(1.0, done + 0.05)
               pz.post(("progress", name, done))
           pz.post(("finished", name))
   
       threading.Thread(target=worker, daemon=True).start()
   
   def update(self, message):
       kind, name, *rest = message
       if kind == "progress":
           self.progress[name] = rest[0]
       elif kind == "finished":
           self.notify(f"{name} downloaded", kind="success")

3. Self-dismissing notifications. A threading.Timer posts the dismiss message:

   def notify(self, title, message=None, kind="info"):
       toast_id = f"{time.monotonic()}"
       self.toasts.append((toast_id, title, message, kind))
       threading.Timer(4, pz.post, args=(("dismiss", toast_id),)).start()
   
   # in view():
   return ui.toasts(body, [ui.toast(t, m, kind=k, on_dismiss=("dismiss", i)) for i, t, m, k in self.toasts])

Note

State is only modified on the UI thread (in update and in callbacks). Worker threads don’t touch self: they send messages with pz.post.

Full code

examples/downloads.py

"""Background work without freezing the UI: Command.perform, pz.post from threads, progress bars and toasts.

    python examples/downloads.py

The "downloads" are simulated with sleep so the example works offline.
Walkthrough: https://albakore.github.io/pyzed/tutorials/intermediate/background-tasks/
"""

import random
import threading
import time

import pyzed as pz
from pyzed import Command, ui

FILES = ["iced-0.14.tar.gz", "zed-themes.json", "ibm-plex.zip", "lucide-icons.ttf"]


class Downloads(pz.App):
    title = "Descargas"
    size = (560, 460)

    def __init__(self):
        self.progress = {}       # name -> 0..1
        self.toasts = []         # (id, title, message, kind)
        self.checking = False
        self.latest = None

    # -- pz.post: a worker thread reports progress through update() -----------------
    def start(self, name):
        self.progress[name] = 0.0

        def worker():
            done = 0.0
            while done < 1.0:
                time.sleep(random.uniform(0.03, 0.12))
                done = min(1.0, done + random.uniform(0.02, 0.08))
                pz.post(("progress", name, done))
            pz.post(("finished", name))

        threading.Thread(target=worker, daemon=True).start()

    def update(self, message):
        kind, name, *rest = message
        if kind == "progress":
            self.progress[name] = rest[0]
        elif kind == "finished":
            self.notify(f"{name} descargado", kind="success")
        elif kind == "dismiss":
            self.toasts = [t for t in self.toasts if t[0] != name]

    def notify(self, title, message=None, kind="info"):
        toast_id = f"{time.monotonic()}"
        self.toasts.append((toast_id, title, message, kind))
        threading.Timer(4, pz.post, args=(("dismiss", toast_id),)).start()

    # -- Command.perform: run a blocking function and get its result back --------------
    def check_updates(self):
        self.checking = True
        return Command.perform(self.fetch_latest_version, self.got_version)

    @staticmethod
    def fetch_latest_version():
        time.sleep(1.5)  # e.g. an HTTP request
        return "0.2.0"

    def got_version(self, version):
        self.checking, self.latest = False, version
        self.notify("Hay una versión nueva", f"Pyzed {version} está disponible.")

    # -- view ----------------------------------------------------------------------------
    def row(self, name):
        value = self.progress.get(name)
        if value is None:
            action = ui.button("Descargar", lambda: self.start(name), size="sm", icon="download")
        elif value < 1:
            action = ui.muted(f"{value:.0%}", size="sm")
        else:
            action = ui.badge("listo", tint="success", icon="check")
        return pz.column([
            pz.row([pz.icon("file-archive", size=16), ui.label(name), pz.hspace(), action], spacing=8, align="center"),
            ui.progress(value or 0, tint="success" if value == 1 else "accent") if value is not None else None,
        ], spacing=6)

    def view(self):
        body = pz.container(
            pz.column([
                pz.row([ui.heading("Descargas", level=2), pz.hspace(),
                        ui.button("Buscar actualizaciones", None if self.checking else self.check_updates,
                                  icon="loader" if self.checking else "refresh-cw", variant="outline")],
                       align="center"),
                ui.card(pz.column([self.row(n) for n in FILES], spacing=14)),
                ui.muted(f"Última versión: {self.latest}" if self.latest else "Pulsa un botón: la interfaz sigue respondiendo.",
                         size="sm"),
            ], spacing=16),
            padding=24,
            width="fill",
            height="fill",
            bg=self.resolved_theme().editor,
        )
        items = [ui.toast(title, msg, kind=kind, on_dismiss=("dismiss", tid)) for tid, title, msg, kind in self.toasts]
        return ui.toasts(body, items)


if __name__ == "__main__":
    pz.run(Downloads)

uv

Windows, macOS and Linux

uv run --with pyzed-ui python examples/downloads.py

pip

macOS / Linux

python3 -m venv .venv
source .venv/bin/activate
python -m pip install pyzed-ui
python examples/downloads.py

Windows (PowerShell)

py -m venv .venv
.venv\Scripts\Activate.ps1
python -m pip install pyzed-ui
python examples\downloads.py

python

macOS / Linux (Pyzed already installed)

python3 examples/downloads.py

Windows (Pyzed already installed)

py examples\downloads.py

Events in this tutorial

This app is declarative: the events of the threads are messages. pz.post(("progress", name, done)) from a worker thread and on_dismiss=("dismiss", tid) of each toast both reach update(message), which unpacks the tuple.

Source	Event	Message or handler
Download button	on_press	lambda: self.start(name)
The worker thread	progress, finished	pz.post(("progress", name, done)), pz.post(("finished", name))
Check for updates	on_press	self.check_updates, which returns Command.perform(work, done)
Command.perform	the work finished	self.got_version(result)
A toast	on_dismiss	the message ("dismiss", tid)

Try more events

def row(self, name):
    ...
    return pz.column([...]).on(double_click=lambda: self.start(name))   # double click: download again

def subscriptions(self):                    # an event every second while something downloads
    busy = any(v < 1 for v in self.progress.values())
    return [pz.every(1, ("tick", "", None))] if busy else []

update gets ("tick", "", None) every second while a download runs, and nothing once they finish: a subscription you stop returning stops.

Next

• A shader effect from the Shader view.
• Connecting a backend: a database behind a repository class.
• Async and subscriptions: coroutines, async generators and streams.