httpx

httpx is the HTTP client a Pyzed window uses to talk to any API: a public one, your own FastAPI or Django server (the client and server architecture). Two demos:

httpx_pypi: concurrent requests to PyPI’s JSON API, shown as they arrive.
httpx_download: a download manager with streaming responses and a progress bar.

The basics, from httpx’s quickstart

httpx has the same API in sync and async form. In a Pyzed app use either:

AsyncClient (recommended)
Sync API in a thread

import httpx

class App(pz.App):
    async def load(self):
        # A coroutine returned from a callback runs on Pyzed's asyncio loop.
        async with httpx.AsyncClient() as client:
            response = await client.get("https://pypi.org/pypi/httpx/json")
        response.raise_for_status()
        return ("loaded", response.json())

    def update(self, message):
        kind, data = message
        self.info = data["info"]

import httpx

def fetch():
    response = httpx.get("https://pypi.org/pypi/httpx/json")
    response.raise_for_status()
    return response.json()

class App(pz.App):
    def load(self):
        # A blocking call runs in a worker thread; then() receives the result.
        return Command.perform(fetch, lambda data: setattr(self, "info", data["info"]))

The rest of the quickstart applies unchanged:

Task	httpx
Query parameters	`client.get(url, params={"q": "pyzed", "page": 2})`
Send JSON	`client.post(url, json={"name": "Deadpond"})`
Send a form or files	`client.post(url, data={...}, files={"upload": open("a.png", "rb")})`
Headers and auth	`httpx.AsyncClient(headers={"Authorization": f"Bearer {token}"})`, `auth=("user", "pass")`
A base URL	`httpx.AsyncClient(base_url="https://api.example.com")` then `client.get("/items/")`
Timeouts	`httpx.AsyncClient(timeout=10)`
Errors	`response.raise_for_status()`; catch `httpx.HTTPStatusError` and `httpx.RequestError`

Concurrent requests: the PyPI explorer

A PyPI explorer listing packages with versions, and the details of FastAPI

cd examples/integrations/httpx_pypi
uv run main.py
uv run main.py rich textual polars      # other packages

cd examples/integrations/httpx_pypi
python3 -m venv .venv
source .venv/bin/activate
pip install pyzed-ui httpx pydantic
python main.py
python main.py rich textual polars      # other packages

cd examples/integrations/httpx_pypi
py -m venv .venv
.venv\Scripts\Activate.ps1
pip install pyzed-ui httpx pydantic
python main.py
python main.py rich textual polars      # other packages

cd examples/integrations/httpx_pypi
python3 -m pip install pyzed-ui httpx pydantic
python3 main.py
python3 main.py rich textual polars      # other packages

cd examples/integrations/httpx_pypi
py -m pip install pyzed-ui httpx pydantic
py main.py
py main.py rich textual polars      # other packages

Return a value for every outcome. The model, or a short text for the list:

async def fetch(client, name):
    try:
        response = await client.get(f"https://pypi.org/pypi/{name}/json")
    except httpx.HTTPError as error:
        return f"Network error: {error}"
    if response.status_code == 404:
        return "Not on PyPI"
    return Package.model_validate_json(response.content)  # see the Pydantic page

Send the requests at once and yield each answer. An async generator returned from on_start or a callback sends every yield to update, so the list fills in as answers arrive:

async def fetch_all(names):
    async with httpx.AsyncClient(timeout=10, follow_redirects=True) as client:
        async def one(name):
            return name, await fetch(client, name)

        for done in asyncio.as_completed([one(n) for n in names]):
            yield await done

class Explorer(pz.App):
    def on_start(self):
        return fetch_all(self.names)

    def update(self, message):
        name, result = message
        self.packages[name] = result

main.py
pypi.py

"""A PyPI explorer: concurrent httpx requests streamed into the window, responses validated by Pydantic.

    uv run main.py
    uv run main.py rich textual polars      # start with other packages

Walkthrough: https://albakore.github.io/pyzed/integrations/httpx/
"""

from __future__ import annotations

import sys
import webbrowser
from datetime import datetime, timezone

import pyzed as pz
from pyzed import ui

from pypi import Package, fetch_all

DEFAULT = ["pyzed-ui", "fastapi", "django", "sqlalchemy", "sqlmodel", "httpx", "pydantic", "requests"]


def ago(when: datetime) -> str:
    days = (datetime.now(timezone.utc) - when).days
    if days < 1:
        return "today"
    if days < 60:
        return f"{days} days ago"
    return f"{days // 30} months ago" if days < 730 else f"{days // 365} years ago"


class Explorer(pz.App):
    title = "PyPI explorer · httpx + Pydantic"
    size = (1100, 680)

    def __init__(self, names: list[str]):
        self.names = names
        self.packages: dict[str, Package | str | None] = {name: None for name in names}  # None while loading
        self.selected = names[0] if names else None
        self.new_name = ""

    def on_start(self):
        return fetch_all(self.names)  # an async generator: each (name, result) it yields reaches update()

    def update(self, message):
        name, result = message
        self.packages[name] = result

    def add(self, *_):
        name = self.new_name.strip().lower()
        self.new_name = ""
        if not name:
            return None
        self.selected = name
        if name in self.packages and isinstance(self.packages[name], Package):
            return None
        self.packages[name] = None
        return fetch_all([name])

    # -- view ----------------------------------------------------------------------------
    def row(self, name: str):
        package = self.packages[name]
        if package is None:
            end = ui.spinner(size=12)
        elif isinstance(package, str):
            end = pz.icon("circle-alert", size=14, color=self.resolved_theme().error)
        else:
            end = ui.muted(package.info.version, size="sm", mono=True)
        return ui.list_item(name, lambda: setattr(self, "selected", name), icon="package", end=end,
                            selected=name == self.selected, inset=True, height=30)

    def details(self, t):
        package = self.packages.get(self.selected)
        if package is None:
            return pz.center(ui.spinner(size=24))
        if isinstance(package, str):
            return pz.center(ui.empty_state(self.selected, package, icon="circle-alert"))
        info, history = package.info, package.history()
        years = package.per_year()
        links = [ui.button(label, lambda url=url: webbrowser.open(url), icon="external-link", size="sm", variant="ghost")
                 for label, url in list((info.project_urls or {}).items())[:4]]
        versions = ui.virtual_list(history, lambda item: pz.row([
            ui.label(item[0], mono=True, width="fill"), ui.muted(item[1].strftime("%Y-%m-%d"), size="sm")],
            padding=(0, 8), align="center", height=26), item_height=26)
        return pz.column([
            pz.row([ui.heading(info.name, level=2), ui.badge(info.version, tint="success")], spacing=10, align="center"),
            ui.muted(info.summary or "No summary.", size="lg"),
            pz.row(links, spacing=4) if links else None,
            pz.row([
                ui.stat("Releases", len(history)),
                ui.stat("Latest", ago(history[0][1]) if history else "—"),
                ui.stat("Python", info.requires_python or "any"),
                ui.stat("License", (info.license_expression or info.license or "—")[:18]),
            ], spacing=12),
            pz.row([
                ui.card(ui.bar_chart([count for _, count in years], labels=[str(y)[2:] for y, _ in years],
                                     width=340, height=150), title="Releases per year", width=374),
                ui.card(pz.container(versions, height=150), title="Versions", width="fill"),
            ], spacing=12),
        ], spacing=14, padding=20)

    def view(self):
        t = self.resolved_theme()
        loading = sum(p is None for p in self.packages.values())
        sidebar = pz.container(pz.column([
            ui.input(self.new_name, lambda v: setattr(self, "new_name", v), placeholder="Add a package…",
                     icon="plus", on_submit=self.add),
            ui.virtual_list(list(self.packages), self.row, item_height=30),
            ui.muted(f"Loading {loading}…" if loading else f"{len(self.packages)} packages", size="sm"),
        ], spacing=8), padding=8, width=260, height="fill", bg=t.panel)
        return pz.row([sidebar, pz.container(self.details(t), width="fill", height="fill", bg=t.editor)])


if __name__ == "__main__":
    pz.run(Explorer(sys.argv[1:] or DEFAULT))

"""PyPI's JSON API, described with Pydantic models and fetched with httpx."""

from __future__ import annotations

import asyncio
from collections import Counter
from datetime import datetime
from typing import AsyncIterator

import httpx
from pydantic import BaseModel, ValidationError


class Info(BaseModel):
    name: str
    version: str
    summary: str | None = None
    license: str | None = None
    license_expression: str | None = None  # newer packages declare an SPDX expression instead
    requires_python: str | None = None
    project_urls: dict[str, str] | None = None


class File(BaseModel):
    upload_time_iso_8601: datetime
    yanked: bool = False


class Package(BaseModel):
    """Only the fields the app shows; Pydantic ignores the rest of the response."""

    info: Info
    releases: dict[str, list[File]]

    def history(self) -> list[tuple[str, datetime]]:
        """(version, first upload) for every release with files, newest first."""
        dated = [(version, min(f.upload_time_iso_8601 for f in files)) for version, files in self.releases.items() if files]
        return sorted(dated, key=lambda item: item[1], reverse=True)

    def per_year(self, years: int = 8) -> list[tuple[int, int]]:
        counts = Counter(date.year for _, date in self.history())
        last = datetime.now().year
        return [(year, counts.get(year, 0)) for year in range(last - years + 1, last + 1)]


async def fetch(client: httpx.AsyncClient, name: str) -> Package | str:
    """The package, or a short message saying why it couldn't be loaded."""
    try:
        response = await client.get(f"https://pypi.org/pypi/{name}/json")
    except httpx.HTTPError as error:
        return f"Network error: {error or type(error).__name__}"
    if response.status_code == 404:
        return "Not on PyPI"
    if response.is_error:
        return f"PyPI answered {response.status_code}"
    try:
        return Package.model_validate_json(response.content)
    except ValidationError as error:
        return f"Unexpected response: {error.error_count()} problems"


async def fetch_all(names: list[str]) -> AsyncIterator[tuple[str, Package | str]]:
    """Requests every package at once and yields each one as soon as it arrives."""
    async with httpx.AsyncClient(timeout=10, follow_redirects=True) as client:
        async def one(name):
            return name, await fetch(client, name)

        for done in asyncio.as_completed([one(n) for n in names]):
            yield await done

Streaming downloads with progress

A download manager with progress bars

cd examples/integrations/httpx_download
uv run main.py

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

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

cd examples/integrations/httpx_download
python3 -m pip install pyzed-ui httpx
python3 main.py

cd examples/integrations/httpx_download
py -m pip install pyzed-ui httpx
py main.py

httpx’s streaming responses read the body in chunks instead of loading it into memory, and its download progress recipe uses response.num_bytes_downloaded. In Pyzed the loop becomes an async generator that yields after every chunk, and the window redraws the bar each time:

async def download(item):
    async with httpx.AsyncClient(follow_redirects=True) as client:
        async with client.stream("GET", item.url) as response:
            response.raise_for_status()
            item.total = int(response.headers["Content-Length"])
            with open(FOLDER / item.name, "wb") as file:
                async for chunk in response.aiter_bytes():
                    if item.cancel:          # set by the Cancel button
                        break
                    file.write(chunk)
                    item.done = response.num_bytes_downloaded
                    yield item               # reaches update(), which redraws

class Downloads(pz.App):
    def start(self, *_):
        item = Download(self.url)
        self.items.insert(0, item)
        return download(item)

    def view(self):
        ...
        ui.progress(item.done / item.total)

Several downloads can run at the same time: each one is its own generator on the same loop.

"""A download manager built on httpx's streaming responses, with a live progress bar.

    uv run main.py

It follows httpx's "Monitoring download progress" recipe
(https://www.python-httpx.org/advanced/clients/#monitoring-download-progress), written with
``AsyncClient.stream`` and ``aiter_bytes``. The download is an async generator: every progress
report it yields reaches ``update``, so the window redraws while the bytes arrive.
Files are saved in a ``downloads`` folder next to this file.
Walkthrough: https://albakore.github.io/pyzed/integrations/httpx/
"""

from __future__ import annotations

import time
from pathlib import Path
from urllib.parse import urlsplit

import httpx

import pyzed as pz
from pyzed import ui

FOLDER = Path(__file__).with_name("downloads")
EXAMPLE = ("https://files.pythonhosted.org/packages/54/a4/f8188c4f3e07f7737683588210c073478abcb542048cf4ab6fedad0b458a/"
           "numpy-2.1.0.tar.gz")


def size(num_bytes: float) -> str:
    for unit in ("B", "KB", "MB", "GB"):
        if num_bytes < 1024 or unit == "GB":
            return f"{num_bytes:.0f} {unit}" if unit == "B" else f"{num_bytes:.1f} {unit}"
        num_bytes /= 1024


class Download:
    """One row of the list. ``total`` is None when the server sends no Content-Length."""

    def __init__(self, url: str):
        self.url = url
        self.name = Path(urlsplit(url).path).name or "download"
        self.done, self.total = 0, None
        self.speed = 0.0
        self.state = "waiting"  # waiting, downloading, finished, cancelled or the error text
        self.cancel = False     # set by the Cancel button, read by the loop below


async def download(item: Download):
    """Streams ``item.url`` to disk, yielding ``item`` after every chunk."""
    FOLDER.mkdir(exist_ok=True)
    started = time.monotonic()
    try:
        async with httpx.AsyncClient(follow_redirects=True, timeout=30) as client:
            async with client.stream("GET", item.url) as response:
                response.raise_for_status()
                length = response.headers.get("Content-Length")
                item.total, item.state = (int(length) if length else None), "downloading"
                with open(FOLDER / item.name, "wb") as file:
                    async for chunk in response.aiter_bytes():
                        if item.cancel:
                            item.state = "cancelled"
                            break
                        file.write(chunk)
                        item.done = response.num_bytes_downloaded
                        item.speed = item.done / max(time.monotonic() - started, 1e-6)
                        yield item
        if item.state == "downloading":
            item.state = "finished"
    except httpx.HTTPStatusError as error:
        item.state = f"The server answered {error.response.status_code}"
    except httpx.HTTPError as error:
        item.state = f"{type(error).__name__}: {error}"
    yield item


class Downloads(pz.App):
    title = "Downloads · httpx"
    size = (760, 480)

    def __init__(self):
        self.url = EXAMPLE
        self.items: list[Download] = []

    def start(self, *_):
        url = self.url.strip()
        if not url.startswith(("http://", "https://")):
            return None
        item = Download(url)
        self.items.insert(0, item)
        return download(item)  # Pyzed runs the generator on its loop and sends each yield to update()

    def update(self, item: Download):
        pass  # the Download object was changed in place; returning from update redraws the window

    def row(self, item: Download):
        t = self.resolved_theme()
        if item.state == "downloading":
            detail = f"{size(item.done)} of {size(item.total) if item.total else '?'} · {size(item.speed)}/s"
            action = ui.icon_button("x", lambda: setattr(item, "cancel", True), tooltip="Cancel")
        elif item.state == "finished":
            detail, action = f"{size(item.done)} · saved in downloads/", None
        else:
            detail, action = item.state.capitalize(), None
        failed = item.state not in ("waiting", "downloading", "finished")
        value = item.done / item.total if item.total else (1.0 if item.state == "finished" else 0.0)
        return ui.card(pz.column([
            pz.row([pz.icon("file-down", size=16, color=t.icon_muted), ui.label(item.name, width="fill"), action],
                   spacing=8, align="center"),
            ui.progress(value, tint="error" if failed else "success" if item.state == "finished" else "accent"),
            ui.muted(detail, size="sm"),
        ], spacing=8), padding=12)

    def view(self):
        t = self.resolved_theme()
        body = pz.scrollable(pz.column([self.row(i) for i in self.items], spacing=8), height="fill") if self.items \
            else ui.empty_state("No downloads", "Paste a URL and press Download.", icon="download")
        return pz.container(pz.column([
            pz.row([ui.input(self.url, lambda v: setattr(self, "url", v), icon="link", on_submit=self.start),
                    ui.button("Download", self.start, icon="download", variant="solid")], spacing=8),
            body,
        ], spacing=14), padding=20, width="fill", height="fill", bg=t.editor)


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