HTTP requests with httpx
Install
bunpy add httpx tenacityBasic GET request
import httpx
response = httpx.get("https://httpbin.org/get")
print(response.status_code) # 200
print(response.json())Pass query parameters as a dict - httpx URL-encodes them automatically:
import httpx
params = {"q": "python", "per_page": 5}
response = httpx.get("https://api.github.com/search/repositories", params=params)
data = response.json()
print(f"Total results: {data['total_count']}")
for repo in data["items"]:
print(repo["full_name"], "-", repo["stargazers_count"], "stars")POST with JSON body
import httpx
payload = {"title": "Hello", "body": "World", "userId": 1}
response = httpx.post("https://jsonplaceholder.typicode.com/posts", json=payload)
print(response.status_code) # 201
print(response.json())Session with base URL and default headers
A Client is the right tool when you make multiple requests to the same host. It reuses the underlying TCP connection and lets you set shared headers once:
import httpx
import os
GITHUB_TOKEN = os.environ.get("GITHUB_TOKEN", "")
with httpx.Client(
base_url="https://api.github.com",
headers={
"Authorization": f"Bearer {GITHUB_TOKEN}",
"Accept": "application/vnd.github+json",
"X-GitHub-Api-Version": "2022-11-28",
},
timeout=10.0,
) as client:
user = client.get("/user").json()
print(f"Logged in as: {user['login']}")
repos = client.get("/user/repos", params={"per_page": 5}).json()
for repo in repos:
print(repo["full_name"])Paginate GitHub API results
GitHub paginates most list endpoints. Keep fetching until there is no next link:
import httpx
import os
def get_all_repos(org: str, token: str) -> list[dict]:
repos: list[dict] = []
url = f"https://api.github.com/orgs/{org}/repos"
params: dict = {"per_page": 100, "type": "public"}
with httpx.Client(
headers={
"Authorization": f"Bearer {token}",
"Accept": "application/vnd.github+json",
},
timeout=15.0,
) as client:
while url:
response = client.get(url, params=params)
response.raise_for_status()
repos.extend(response.json())
# Follow the Link header for the next page
link = response.headers.get("link", "")
next_url = None
for part in link.split(","):
if 'rel="next"' in part:
next_url = part.split(";")[0].strip().strip("<>")
url = next_url
params = {} # next URL already has query string baked in
return repos
token = os.environ["GITHUB_TOKEN"]
repos = get_all_repos("python", token)
print(f"Found {len(repos)} repos")Handle rate limits
The GitHub API returns a Retry-After or X-RateLimit-Reset header when you are throttled. Check for 429 and 403 and back off:
import httpx
import time
import os
def github_get(client: httpx.Client, url: str, **kwargs) -> httpx.Response:
while True:
response = client.get(url, **kwargs)
if response.status_code == 403:
reset_at = int(response.headers.get("X-RateLimit-Reset", 0))
sleep_for = max(reset_at - int(time.time()), 1)
print(f"Rate limited. Sleeping {sleep_for}s")
time.sleep(sleep_for)
continue
if response.status_code == 429:
retry_after = int(response.headers.get("Retry-After", 5))
print(f"Too many requests. Sleeping {retry_after}s")
time.sleep(retry_after)
continue
response.raise_for_status()
return response
with httpx.Client(
headers={"Authorization": f"Bearer {os.environ['GITHUB_TOKEN']}"},
timeout=10.0,
) as client:
data = github_get(client, "https://api.github.com/repos/python/cpython").json()
print(data["stargazers_count"])Retry with tenacity
For transient network failures - connection resets, 500 errors - wrap requests with tenacity:
import httpx
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
@retry(
retry=retry_if_exception_type((httpx.TransportError, httpx.HTTPStatusError)),
stop=stop_after_attempt(5),
wait=wait_exponential(multiplier=1, min=1, max=30),
)
def fetch_with_retry(url: str) -> dict:
response = httpx.get(url, timeout=10.0)
response.raise_for_status()
return response.json()
data = fetch_with_retry("https://api.github.com/repos/python/cpython")
print(data["name"])Upload a file
import httpx
with open("report.csv", "rb") as f:
response = httpx.post(
"https://httpbin.org/post",
files={"file": ("report.csv", f, "text/csv")},
data={"description": "Monthly sales report"},
)
print(response.status_code)
print(response.json()["files"])Custom authentication
Subclass httpx.Auth to attach credentials to every request:
import httpx
class APIKeyAuth(httpx.Auth):
def __init__(self, api_key: str) -> None:
self.api_key = api_key
def auth_flow(self, request: httpx.Request):
request.headers["X-Api-Key"] = self.api_key
yield request
with httpx.Client(auth=APIKeyAuth("secret-key-here")) as client:
response = client.get("https://httpbin.org/headers")
print(response.json()["headers"].get("X-Api-Key"))Streaming response
Stream large downloads instead of loading the whole body into memory:
import httpx
url = "https://speed.hetzner.de/100MB.bin"
with httpx.stream("GET", url, timeout=60.0) as response:
response.raise_for_status()
total = int(response.headers.get("content-length", 0))
downloaded = 0
with open("large_file.bin", "wb") as f:
for chunk in response.iter_bytes(chunk_size=65536):
f.write(chunk)
downloaded += len(chunk)
if total:
pct = downloaded / total * 100
print(f"\r{pct:.1f}%", end="", flush=True)
print("\nDone.")Async client
Use httpx.AsyncClient inside an async def function. This is the right choice when you are already in an async context (FastAPI, asyncio scripts, etc.):
import asyncio
import httpx
async def fetch_repos(usernames: list[str]) -> dict[str, int]:
results: dict[str, int] = {}
async with httpx.AsyncClient(
base_url="https://api.github.com",
timeout=10.0,
) as client:
tasks = [client.get(f"/users/{u}/repos", params={"per_page": 100}) for u in usernames]
responses = await asyncio.gather(*tasks)
for username, response in zip(usernames, responses):
response.raise_for_status()
results[username] = len(response.json())
return results
counts = asyncio.run(fetch_repos(["torvalds", "gvanrossum", "antirez"]))
for user, count in counts.items():
print(f"{user}: {count} repos")Async streaming
import asyncio
import httpx
async def stream_download(url: str, dest: str) -> None:
async with httpx.AsyncClient(timeout=60.0) as client:
async with client.stream("GET", url) as response:
response.raise_for_status()
with open(dest, "wb") as f:
async for chunk in response.aiter_bytes(chunk_size=65536):
f.write(chunk)
asyncio.run(stream_download("https://httpbin.org/bytes/1024", "output.bin"))
print("Downloaded.")Timeout configuration
Set separate timeouts for connect, read, write, and pool:
import httpx
timeout = httpx.Timeout(
connect=5.0, # time to establish TCP connection
read=30.0, # time to receive data
write=10.0, # time to send request body
pool=5.0, # time to acquire a connection from the pool
)
with httpx.Client(timeout=timeout) as client:
response = client.get("https://httpbin.org/delay/2")
print(response.status_code)Run the examples
# Set your token once
export GITHUB_TOKEN=ghp_...
# Run any script directly
bunpy github_repos.pyhttpx keeps the same familiar requests-style API for synchronous code while offering a true async client. The session (Client) approach is almost always the right default: it handles connection pooling, keeps headers DRY, and makes retry wrappers straightforward to add.