Conectar un backend

Pyzed es el frontend: dibuja ventanas y reacciona al usuario. Tus datos viven en otro sitio: una base de datos, una API REST, archivos. Este tutorial conecta las dos partes como lo hacen las apps grandes, con una capa de backend delgada con la que habla la ventana, y cada llamada lenta fuera del hilo de la interfaz para que la ventana nunca se congele.

Objetivo	Tiempo	Nivel	Necesitas
Una app de contactos sobre SQLite	40 minutos	Avanzado · Python	Tareas en segundo plano

Qué vas a aprender

• Poner todo el acceso a datos en una clase repositorio sin interfaz, fácil de probar y de reemplazar.
• Ejecutar consultas en un hilo en segundo plano con Command.perform(work, then).
• Mostrar un spinner mientras corre una consulta, y convertir las excepciones en un mensaje en lugar de un cierre.
• Recargar la vista después de un cambio (añadir, borrar, buscar).

Cada consulta corre en un hilo; la ventana sigue respondiendo incluso con una base de datos lenta.

La forma de la app

flowchart LR
    UI["Contacts (pz.App)<br/>estado + view()"] -- "Command.perform(work)" --> T["hilo en segundo plano"]
    T -- "work(): repo.search(...)" --> R["ContactsRepository"]
    R --> DB[("SQLite<br/>(o una API, un ORM...)")]
    T -- "then(resultado) en el hilo de la interfaz" --> UI

Pasos

1. El repositorio. Python normal, sin importar Pyzed. Abre una conexión por llamada, así se puede usar desde cualquier hilo:

   class ContactsRepository:
       def __init__(self, path=DEFAULT_DB):
           self.path = Path(path)
           with self.connect() as db:
               db.execute("CREATE TABLE IF NOT EXISTS contacts (id INTEGER PRIMARY KEY, name TEXT, email TEXT)")
   
       def connect(self):
           return sqlite3.connect(self.path)
   
       def search(self, text=""):
           with self.connect() as db:
               rows = db.execute("SELECT id, name, email FROM contacts WHERE name LIKE ? ORDER BY name",
                                 (f"%{text}%",)).fetchall()
           return [{"id": i, "name": n, "email": e} for i, n, e in rows]

2. Nunca consultes en el hilo de la interfaz. Una consulta de 300 ms congelaría la ventana 300 ms. Command.perform(work, then) ejecuta work() en un hilo y llama a then(resultado) de vuelta en el hilo de la interfaz, donde es seguro cambiar el estado:

   def reload(self):
       return pz.Command.perform(lambda: self.repo.search(self.query), self.loaded)
   
   def loaded(self, contacts):
       self.contacts = contacts

3. Ocupado y errores. Una pequeña función envuelve cada llamada: marca self.busy (la vista muestra un spinner) y atrapa las excepciones para que se conviertan en un aviso:

   def background(self, work, then=None):
       self.busy, self.error = True, None
   
       def guarded():
           try:
               return ("ok", work())
           except Exception as exc:
               return ("error", str(exc))
   
       def done(outcome):
           self.busy = False
           kind, value = outcome
           if kind == "error":
               self.error = value
           elif then is not None:
               return then(value)
   
       return pz.Command.perform(guarded, done)

4. Carga al empezar, recarga después de cambios. on_start se ejecuta una vez al abrir la ventana. Añadir o borrar recarga la lista al terminar:

   def on_start(self):
       return self.reload()
   
   def add(self):
       name, email = self.name, self.email
       self.name = self.email = ""
       return self.background(lambda: self.repo.add(name, email), lambda _: self.reload())

5. Prueba el backend solo. Como el repositorio no tiene interfaz, una prueba son tres líneas:

   repo = ContactsRepository(tmp_path / "contacts.db")
   repo.add("Grace Hopper", "grace@example.com")
   assert [c["name"] for c in repo.search("grace")] == ["Grace Hopper"]

Bibliotecas async

Si tu cliente es async (httpx.AsyncClient, aiosqlite, la AsyncSession de SQLAlchemy), pasa un async def a Command.perform, o devuelve una corrutina desde un callback: corre en el loop de asyncio de Pyzed. Mira Async y suscripciones.

No toques self desde el hilo

work() corre en otro hilo: deja que devuelva un valor y cambia self en then. Cambiar el estado desde el hilo de trabajo puede chocar con el dibujado.

Código completo

examples/tutorial/contacts_backend.py

"""Connecting a backend: a contacts app whose data lives in SQLite, behind a small repository class.

    python examples/tutorial/contacts_backend.py

- ``ContactsRepository`` is the backend: plain Python with no UI in it, easy to test and to swap
  for SQLAlchemy, an HTTP API or anything else.
- The window never talks to the database directly: ``Command.perform(work, then)`` runs the query
  in a background thread and hands the result to ``then`` on the UI thread, so the window never
  freezes, even on a slow disk or network.
- A spinner shows while a query runs; errors become a callout instead of a crash.

Walkthrough: https://albakore.github.io/pyzed/tutorials/advanced/backend/
"""

import sqlite3
import tempfile
import time
from pathlib import Path

import pyzed as pz
from pyzed import ui

DEFAULT_DB = Path(tempfile.gettempdir()) / "pyzed-tutorial-contacts.db"


class ContactsRepository:
    """The backend. One connection per call, so it is safe to use from any thread."""

    def __init__(self, path=DEFAULT_DB, delay=0.3):
        self.path = Path(path)
        self.delay = delay  # pretend the database is far away, to see the spinner
        with self.connect() as db:
            db.execute("CREATE TABLE IF NOT EXISTS contacts (id INTEGER PRIMARY KEY, name TEXT, email TEXT)")
            if not db.execute("SELECT COUNT(*) FROM contacts").fetchone()[0]:
                db.executemany("INSERT INTO contacts (name, email) VALUES (?, ?)",
                               [("Ada Lovelace", "ada@example.com"), ("Alan Turing", "alan@example.com")])

    def connect(self):
        return sqlite3.connect(self.path)

    def search(self, text=""):
        time.sleep(self.delay)
        with self.connect() as db:
            rows = db.execute("SELECT id, name, email FROM contacts WHERE name LIKE ? OR email LIKE ? ORDER BY name",
                              (f"%{text}%", f"%{text}%")).fetchall()
        return [{"id": i, "name": n, "email": e} for i, n, e in rows]

    def add(self, name, email):
        if not name.strip() or "@" not in email:
            raise ValueError("A contact needs a name and a valid email.")
        time.sleep(self.delay)
        with self.connect() as db:
            db.execute("INSERT INTO contacts (name, email) VALUES (?, ?)", (name.strip(), email.strip()))

    def delete(self, contact_id):
        with self.connect() as db:
            db.execute("DELETE FROM contacts WHERE id = ?", (contact_id,))


class Contacts(pz.App):
    title = "Contacts"
    size = (640, 520)

    def __init__(self, repository=None):
        self.repo = repository or ContactsRepository()
        self.contacts = []
        self.query = ""
        self.name = self.email = ""
        self.busy = False
        self.error = None

    # -- talking to the backend ---------------------------------------------------------
    def background(self, work, then=None):
        """Runs ``work()`` in a thread; ``then(result)`` gets the result on the UI thread."""
        self.busy, self.error = True, None

        def guarded():
            try:
                return ("ok", work())
            except Exception as exc:  # noqa: BLE001 - show the problem instead of crashing
                return ("error", str(exc))

        def done(outcome):
            self.busy = False
            kind, value = outcome
            if kind == "error":
                self.error = value
            elif then is not None:
                return then(value)

        return pz.Command.perform(guarded, done)

    def on_start(self):
        return self.reload()

    def reload(self):
        return self.background(lambda: self.repo.search(self.query), self.loaded)

    def loaded(self, contacts):
        self.contacts = contacts

    def search(self, text):
        self.query = text
        return self.reload()

    def add(self):
        name, email = self.name, self.email
        self.name = self.email = ""
        return self.background(lambda: self.repo.add(name, email), lambda _: self.reload())

    def delete(self, contact):
        return self.background(lambda: self.repo.delete(contact["id"]), lambda _: self.reload())

    # -- view ---------------------------------------------------------------------------
    def row(self, contact):
        initials = "".join(part[0] for part in contact["name"].split()[:2])
        return ui.list_item(
            pz.row([ui.label(contact["name"], weight="medium"), ui.muted(contact["email"])], spacing=10),
            None, start=ui.avatar(initials, size=26, index=contact["id"]), height=40,
            end=ui.icon_button("trash-2", lambda: self.delete(contact), size="sm", tooltip="Delete"),
        )

    def view(self):
        t = self.resolved_theme()
        form = pz.row([
            ui.input(self.name, lambda v: setattr(self, "name", v), placeholder="Name"),
            ui.input(self.email, lambda v: setattr(self, "email", v), placeholder="Email", on_submit=self.add),
            ui.button("Add", self.add, variant="solid", icon="plus"),
        ], spacing=8)
        body = (ui.empty_state("No contacts", "Add one above.", icon="users") if not self.contacts else
                pz.scrollable(pz.column([self.row(c) for c in self.contacts], spacing=2), height="fill"))
        return pz.container(pz.column([
            pz.row([ui.heading("Contacts", level=2), pz.hspace(), ui.spinner() if self.busy else None],
                   align="center"),
            ui.search_input(self.query, self.search, placeholder="Search by name or email…"),
            form,
            ui.callout("Could not save", self.error, tint="error") if self.error else None,
            body,
        ], spacing=12), padding=24, width="fill", height="fill", bg=t.background)


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

uv

Windows, macOS y Linux

uv run --with pyzed-ui python examples/tutorial/contacts_backend.py

pip

macOS / Linux

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

Windows (PowerShell)

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

python

macOS / Linux (con Pyzed ya instalado)

python3 examples/tutorial/contacts_backend.py

Windows (con Pyzed ya instalado)

py examples\tutorial\contacts_backend.py

Backends reales

El mismo patrón con bibliotecas reales, cada uno un proyecto completo en examples/integrations:

SQLAlchemy→Sesiones en hilos o una AsyncSession, y 20.000 filas en una tabla virtual.FastAPI→Un cliente de escritorio para una API: servidor aparte, embebido o en el mismo proceso.Django→El tutorial de encuestas manejado desde una ventana con el ORM o por HTTP.httpx→Peticiones concurrentes y descargas en streaming con progreso.

Eventos de este tutorial

Origen	Evento	Handler
La app	on_start	Carga los contactos del backend en segundo plano.
ui.input	on_input	Guarda el texto de cada campo.
ui.input (email)	on_submit	self.add: manda el contacto nuevo al backend.

Prueba más eventos

def row(self, contact):
    return ui.list_item(contact.name, None, icon="user",
                        on_double_press=lambda: setattr(self, "editing", contact),  # editarlo
                        on_right_press=lambda: setattr(self, "deleting", contact))  # pedir borrarlo

def on_key(self, combo):
    if combo == "f5":                       # recargar desde el backend
        return self.on_start()

Siguiente

• Empaqueta tu app para Windows, macOS y Linux.
• Integraciones: dónde corre tu código y cómo los formularios muestran errores del servidor.