Fetch

import json

from pyxpo_fetch import fetch


async def add_item(token: str) -> dict:
    resp = await fetch(
        "https://api.example.com/items",
        method="POST",
        headers={"Content-Type": "application/json", "Authorization": "Bearer " + token},
        body=json.dumps({"name": "pen"}),
    )
    return await resp.json()

import asyncio

from pyxpo_fetch import AbortController, AbortError, fetch


async def read_events(handle):
    resp = await fetch("https://example.com/events")
    async for chunk in resp.body:
        handle(chunk)          # bytes, as they arrive


async def slow_call(url):
    ctl = AbortController()
    task = asyncio.ensure_future(fetch(url, signal=ctl.signal))
    await asyncio.sleep(5)
    ctl.abort()
    try:
        return await task
    except AbortError:
        return None

from urllib.parse import quote

from pyxpo import Text, TextInput, View, component, use_async, use_state
from pyxpo_fetch import FetchError, fetch


@component
def Search():
    q, set_q = use_state("")
    result, set_result = use_state("")

    async def look_up():
        if len(q) < 2:
            set_result("")
            return
        try:
            resp = await fetch("https://api.example.com/search?q=" + quote(q), timeout=8)
            if not resp.ok:
                set_result("Server said %d" % resp.status)
                return
            data = await resp.json()
            set_result(", ".join(item["name"] for item in data.get("items", [])))
        except FetchError as e:
            set_result("Could not search: %s" % e)

    use_async(look_up, [q])
    return View(TextInput(value=q, on_change_text=set_q, placeholder="Search"), Text(result))

import asyncio

import pytest
import pyxpo_fetch as mod
from pyxpo_fetch import Headers


def test_headers_case_insensitive():
    h = Headers({"Content-Type": "text/plain"})
    h.append("Set-Cookie", "a=1")
    h.append("set-cookie", "b=2")
    assert h.get("content-type") == "text/plain"
    assert h.get_set_cookie() == ["a=1", "b=2"]


def test_validation():
    with pytest.raises(ValueError):
        asyncio.run(mod.fetch("ftp://x.test/"))
    with pytest.raises(ValueError):
        asyncio.run(mod.fetch("https://x.test/", method="GET", body="x"))