# httpx

> Call web APIs from a Pyzed app with httpx, following its quickstart: requests, an AsyncClient, concurrent calls, errors and streaming downloads with a progress bar.

Source: https://albakore.github.io/pyzed/integrations/httpx/

[httpx](https://www.python-httpx.org/) is the HTTP client a Pyzed window uses to talk to any API:
a public one, your own [FastAPI](https://albakore.github.io/pyzed/integrations/fastapi/) or [Django](https://albakore.github.io/pyzed/integrations/django/)
server (the client and server [architecture](https://albakore.github.io/pyzed/integrations/architectures/)). 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):**

```python
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"]
```

**Sync API in a thread:**

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

**Tip:**

Create one `AsyncClient` and reuse it, as [httpx recommends](https://www.python-httpx.org/advanced/clients/),
to keep connections open. All of Pyzed's coroutines share one asyncio loop, so a client created
inside a coroutine can be stored on `self` and used by later ones (the FastAPI demo does this).

## Concurrent requests: the PyPI explorer

(Image: A PyPI explorer listing packages with versions, and the details of FastAPI)

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

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

2. **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:

```python
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:**
```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.py:**
```py
"""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

(Image: A download manager with progress bars)

httpx's [streaming responses](https://www.python-httpx.org/quickstart/#streaming-responses) read
the body in chunks instead of loading it into memory, and its
[download progress](https://www.python-httpx.org/advanced/clients/#monitoring-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:

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

`examples/integrations/httpx_download/main.py`:

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