Other libraries

Every library falls into one of the cases in the overview: async, blocking, a server or a stream. These recipes cover the ones people ask about most.

aiosqlite

aiosqlite gives SQLite an async API. Because Pyzed runs every coroutine on one loop, a single connection opened in on_start can serve the whole app:

A journal app storing entries with aiosqlite

class Journal(pz.App):
    async def on_start(self):
        self.db = await aiosqlite.connect("journal.db")
        self.db.row_factory = aiosqlite.Row
        await self.db.execute("CREATE TABLE IF NOT EXISTS entries (...)")
        return await self.load()

    async def load(self):
        async with self.db.execute("SELECT * FROM entries ORDER BY id DESC") as cursor:
            return ("loaded", await cursor.fetchall())

cd examples/integrations/aiosqlite_journal
uv run main.py

cd examples/integrations/aiosqlite_journal
python3 -m venv .venv
source .venv/bin/activate
pip install pyzed-ui aiosqlite
python main.py

cd examples/integrations/aiosqlite_journal
py -m venv .venv
.venv\Scripts\Activate.ps1
pip install pyzed-ui aiosqlite
python main.py

cd examples/integrations/aiosqlite_journal
python3 -m pip install pyzed-ui aiosqlite
python3 main.py

cd examples/integrations/aiosqlite_journal
py -m pip install pyzed-ui aiosqlite
py main.py

Full code: examples/integrations/aiosqlite_journal/main.py

"""A journal on aiosqlite: one async SQLite connection living on Pyzed's asyncio loop.

    uv run main.py

The database (journal.db) is created next to this file on the first run.
Walkthrough: https://albakore.github.io/pyzed/integrations/other-libraries/#aiosqlite
"""

from __future__ import annotations

from datetime import datetime
from pathlib import Path

import aiosqlite

import pyzed as pz
from pyzed import ui

DATABASE = Path(__file__).with_name("journal.db")
MOODS = {"great": ("smile", "success"), "okay": ("meh", "warning"), "rough": ("frown", "error")}


class Journal(pz.App):
    title = "Journal · aiosqlite"
    size = (720, 560)

    def __init__(self):
        self.db: aiosqlite.Connection | None = None  # opened on Pyzed's loop, used only from coroutines
        self.entries: list[aiosqlite.Row] = []
        self.text = ""
        self.mood = "great"

    async def on_start(self):
        self.db = await aiosqlite.connect(DATABASE)
        self.db.row_factory = aiosqlite.Row
        await self.db.execute("""CREATE TABLE IF NOT EXISTS entries (
            id INTEGER PRIMARY KEY, written TEXT NOT NULL, mood TEXT NOT NULL, text TEXT NOT NULL)""")
        await self.db.commit()
        return await self.load()

    async def load(self):
        async with self.db.execute("SELECT * FROM entries ORDER BY id DESC LIMIT 200") as cursor:
            return ("loaded", await cursor.fetchall())

    async def add(self, text: str, mood: str):
        await self.db.execute("INSERT INTO entries (written, mood, text) VALUES (?, ?, ?)",
                              (datetime.now().isoformat(timespec="minutes"), mood, text))
        await self.db.commit()
        return await self.load()

    async def remove(self, entry_id: int):
        await self.db.execute("DELETE FROM entries WHERE id = ?", (entry_id,))
        await self.db.commit()
        return await self.load()

    def submit(self, *_):
        text, self.text = self.text.strip(), ""
        return self.add(text, self.mood) if text else None

    def update(self, message):
        kind, rows = message
        if kind == "loaded":
            self.entries = rows

    def entry(self, row):
        icon, tint = MOODS[row["mood"]]
        return ui.card(pz.row([
            pz.icon(icon, size=18, color=self.resolved_theme().color(tint)),
            pz.column([ui.label(row["text"]), ui.muted(row["written"].replace("T", " · "), size="sm")],
                      spacing=4, width="fill"),
            ui.icon_button("trash-2", lambda: self.remove(row["id"]), size="sm", tooltip="Delete"),
        ], spacing=12, align="center"), padding=12)

    def view(self):
        t = self.resolved_theme()
        composer = pz.row([
            ui.toggle_group([(mood, icon) for mood, (icon, _) in MOODS.items()], self.mood, lambda m: setattr(self, "mood", m)),
            ui.input(self.text, lambda v: setattr(self, "text", v), placeholder="How was your day? (Enter to save)",
                     on_submit=self.submit),
        ], spacing=8, align="center")
        entries = pz.scrollable(pz.column([self.entry(r) for r in self.entries], spacing=8), height="fill") \
            if self.entries else ui.empty_state("Nothing written yet", "Your entries are stored in journal.db.",
                                                icon="notebook-pen")
        return pz.container(pz.column([ui.heading("Journal", level=2), composer, entries], spacing=14),
                            padding=20, width="fill", height="fill", bg=t.editor)


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

SQLAlchemy asyncio

Covered on the SQLAlchemy page: the sqlalchemy_quickstart demo runs the same window on a sync Session or on an AsyncSession with aiosqlite (uv run main.py --async).

pandas and Polars

ui.table takes a list of dicts, which is one call away from a DataFrame. Read big files in a thread so the window stays responsive:

import pandas as pd


class Sales(pz.App):
    def __init__(self):
        self.frame = pd.DataFrame()

    def open(self, path):
        return pz.Command.perform(lambda: pd.read_csv(path), self.loaded)

    def loaded(self, frame):
        self.frame = frame
        self.rows = frame.to_dict("records")       # Polars: frame.to_dicts()

    def view(self):
        columns = [ui.Column(name, name) for name in self.frame.columns]
        return ui.table(columns, self.rows, height="fill")

The table is virtualized, so hundreds of thousands of rows scroll smoothly. For filtering, run the query on the DataFrame (frame[frame.city == city]) and rebuild self.rows.

requests and other blocking clients

Any blocking call goes through Command.perform. raise_for_status errors are worth catching in the function so then always gets something to show:

import requests


def fetch_weather(city):
    try:
        response = requests.get("https://wttr.in/" + city, params={"format": "j1"}, timeout=10)
        response.raise_for_status()
        return response.json()
    except requests.RequestException as error:
        return {"error": str(error)}


class Weather(pz.App):
    def search(self, city):
        return pz.Command.perform(lambda: fetch_weather(city), self.show)

Redis pub/sub and other streams

A stream of events is a subscription: pz.listen runs an async generator while it appears in subscriptions() and stops it when it no longer does:

import redis.asyncio as redis


async def events():
    client = redis.Redis()
    async with client.pubsub() as pubsub:
        await pubsub.subscribe("orders")
        async for message in pubsub.listen():
            if message["type"] == "message":
                yield ("order", message["data"].decode())


class Orders(pz.App):
    def subscriptions(self):
        return [pz.listen(events)] if self.live else []

WebSockets work the same way; see examples/websocket_stream.py and the Async guide.

Matplotlib

Pyzed draws its own charts (ui.bar_chart, ui.sparkline, ui.donut, pz.canvas), but a Matplotlib figure can be shown as an image. pz.image accepts PNG bytes; build the figure with matplotlib.figure.Figure (not pyplot) so it is safe to render in a thread:

import io
from matplotlib.figure import Figure


def render(values) -> bytes:
    figure = Figure(figsize=(6, 3), dpi=144)
    figure.subplots().plot(values)
    buffer = io.BytesIO()
    figure.savefig(buffer, format="png", transparent=True)
    return buffer.getvalue()


class Plot(pz.App):
    def refresh(self):
        return pz.Command.perform(lambda: render(self.values), lambda png: setattr(self, "png", png))

    def view(self):
        return pz.image(self.png) if self.png else ui.spinner()