HTTPX Proxy Tutorial for Async Python

import asyncio
import itertools
import httpx

URLS = [
    "https://example.com/page/1",
    "https://example.com/page/2",
]
PROXIES = [
    "http://USER:PASS@gate-1.example:8000",
    "http://USER:PASS@gate-2.example:8000",
]

async def fetch(client, url, semaphore):
    async with semaphore:
        try:
            r = await client.get(url)
            r.raise_for_status()
            return {"url": url, "status": r.status_code, "bytes": len(r.content)}
        except httpx.HTTPError as exc:
            return {"url": url, "error": type(exc).__name__}

async def main():
    semaphore = asyncio.Semaphore(8)
    clients = [
        httpx.AsyncClient(proxy=p, timeout=20, follow_redirects=True)
        for p in PROXIES
    ]
    try:
        tasks = [
            fetch(client, url, semaphore)
            for url, client in zip(URLS, itertools.cycle(clients))
        ]
        print(await asyncio.gather(*tasks))
    finally:
        await asyncio.gather(*(client.aclose() for client in clients))

asyncio.run(main())

[{'url': 'https://example.com/page/1', 'status': 200, 'bytes': 12480},
 {'url': 'https://example.com/page/2', 'error': 'ReadTimeout'}]

Proxies With HTTPX: Async Python Tutorial

TL;DR

Configure an HTTPX Proxy

Most Popular Articles

Proxies With HTTPX: Async Python Tutorial

Nstdata Python SDK Quickstart: Your First Crawl in 5 Minutes

Rotate Proxies Without Breaking Connection Pools

Retry Only What May Recover

Common Failures

Final Verdict

Experience Nstdata — Start Your Free Trial Today

FAQ