# Other libraries

> Recipes for aiosqlite, pandas, Polars, requests, Redis and Matplotlib in a Pyzed desktop app.

Source: https://albakore.github.io/pyzed/integrations/other-libraries/

Every library falls into one of the cases in the [overview](https://albakore.github.io/pyzed/integrations/overview/#where-your-code-runs):
async, blocking, a server or a stream. These recipes cover the ones people ask about most.

## aiosqlite

[aiosqlite](https://aiosqlite.omnilib.dev/) 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:

(Image: A journal app storing entries with aiosqlite)

```python
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())
```

<details>
<summary>Full code: <code>examples/integrations/aiosqlite_journal/main.py</code></summary>

```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)
```

</details>

## SQLAlchemy asyncio

Covered on the [SQLAlchemy page](https://albakore.github.io/pyzed/integrations/sqlalchemy/#the-asyncio-version): 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:

```python
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:

```python
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:

```python
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](https://albakore.github.io/pyzed/guides/async/#subscriptions-timers-and-streams).

## 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:

```python
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()
```