Using proxies in Python: requests, httpx and aiohttp

All three mainstream Python HTTP clients support proxies, and all three have a configuration that looks correct but quietly costs you performance. This guide gives the working setup for each, then covers the parts that actually determine whether your scraper survives.

#requests

import requests

proxies = {
    "http":  "http://user:[email protected]:8000",
    "https": "http://user:[email protected]:8000",
}

# Session, not requests.get — see below
with requests.Session() as s:
    s.proxies.update(proxies)
    r = s.get("https://api.example.com/v1/items", timeout=(5, 30))
    print(r.status_code, r.json())

Note that the https key still points at an http:// proxy URL. That is correct: the scheme in the key is the destination protocol, the scheme in the value is how you reach the proxy. HTTPS requests will use a CONNECT tunnel.

The timeout is a tuple: connect timeout, then read timeout. A bare timeout=30 applies to both and is usually too generous for the connect phase.

#SOCKS5 with requests

pip install "requests[socks]"

proxies = {
    "http":  "socks5h://user:[email protected]:1080",
    "https": "socks5h://user:[email protected]:1080",
}

Use socks5h, not socks5. The h makes the proxy resolve the hostname; without it your machine resolves locally, which leaks DNS and can send you to the wrong regional endpoint. This is covered in more detail in the SOCKS5 entry.

#httpx

import httpx

with httpx.Client(
    proxy="http://user:[email protected]:8000",
    timeout=httpx.Timeout(30.0, connect=5.0),
    follow_redirects=True,
) as client:
    r = client.get("https://api.example.com/v1/items")
    print(r.status_code, r.json())

#Async

import asyncio, httpx

async def fetch_all(urls, limit=10):
    sem = asyncio.Semaphore(limit)          # cap concurrency yourself
    async with httpx.AsyncClient(
        proxy="http://user:[email protected]:8000",
        timeout=httpx.Timeout(30.0, connect=5.0),
    ) as client:
        async def one(u):
            async with sem:
                try:
                    r = await client.get(u)
                    return u, r.status_code, len(r.content)
                except httpx.HTTPError as e:
                    return u, None, repr(e)
        return await asyncio.gather(*(one(u) for u in urls))

print(asyncio.run(fetch_all(["https://api.example.com/v1/items"] * 5)))

#aiohttp

import aiohttp, asyncio

async def main():
    timeout = aiohttp.ClientTimeout(total=30, connect=5)
    async with aiohttp.ClientSession(timeout=timeout) as session:
        async with session.get(
            "https://api.example.com/v1/items",
            proxy="http://gateway.example:8000",
            proxy_auth=aiohttp.BasicAuth("user", "pass"),
        ) as r:
            print(r.status, await r.json())

asyncio.run(main())

aiohttp differs from the others: the proxy is passed per request, not on the session, and credentials go in proxy_auth rather than the URL. aiohttp does not support SOCKS natively — use the aiohttp-socks package if you need it.

#The mistake that costs the most

Calling requests.get() directly in a loop. Each call builds a new connection pool, performs a fresh TCP handshake and a fresh TLS handshake through the proxy, then discards it all.

On a per-gigabyte plan the handshakes are also billable traffic.

#Retries that do not make things worse

import httpx, random, time

def get_with_retry(client, url, attempts=4):
    for i in range(attempts):
        try:
            r = client.get(url)
            if r.status_code == 429:                       # rate limited
                wait = float(r.headers.get("Retry-After", 2 ** i))
            elif r.status_code >= 500:
                wait = 2 ** i
            else:
                return r
        except httpx.HTTPError:
            wait = 2 ** i
        time.sleep(wait + random.uniform(0, 0.5))          # jitter matters
    return None

Two details are load-bearing. Honour Retry-After when the server sends it. And add jitter, because without it parallel workers that fail together retry together and rebuild the burst that triggered rate limiting in the first place.

#The limitation neither library can fix

requests and httpx produce their own TLS fingerprint. A destination can identify the client library from the handshake alone, before it sees a single header you set. Setting a browser User-Agent on top of a Python handshake creates an inconsistency that is more conspicuous than leaving the default.

If your target fingerprints TLS, no amount of header work in these libraries will help. The options are a client that reproduces a browser handshake, or driving a real headless browser. Recognising which problem you have saves days.

#Validate the response, not the status

r = client.get(url)
if r.status_code == 200 and "expected-marker" in r.text:
    handle(r)
else:
    record_failure(url, r.status_code, len(r.content))

Many anti-bot systems return 200 with a challenge page or an empty result set, precisely because it wastes a scraper’s time. A pipeline that checks only the status code will report a high success rate while collecting nothing.

#Rotating through a list yourself

Many providers give a backconnect gateway that rotates for you. When you hold a list of individual addresses instead, rotate deliberately rather than randomly, and remove endpoints that are failing:

import itertools, threading, httpx

class ProxyRing:
    def __init__(self, proxies):
        self._all = list(proxies)
        self._live = list(proxies)
        self._cycle = itertools.cycle(self._live)
        self._lock = threading.Lock()

    def next(self):
        with self._lock:
            return next(self._cycle)

    def drop(self, proxy):
        """Retire an endpoint that keeps failing, and rebuild the cycle."""
        with self._lock:
            if proxy in self._live and len(self._live) > 1:
                self._live.remove(proxy)
                self._cycle = itertools.cycle(self._live)

ring = ProxyRing([
    "http://user:[email protected]:8000",
    "http://user:[email protected]:8000",
])

def fetch(url):
    proxy = ring.next()
    try:
        with httpx.Client(proxy=proxy, timeout=20) as c:
            r = c.get(url)
            if r.status_code in (403, 407):
                ring.drop(proxy)
            return r
    except httpx.HTTPError:
        ring.drop(proxy)
        return None

Dropping on 407 matters: that is the proxy refusing you, so retrying the same endpoint will fail identically. Dropping on 403 is a judgement call, since the destination may be refusing the address specifically.

#Keeping a session on one address

When a flow spans several requests, hold the address for the whole flow:

import httpx, uuid

def flow_client(base_user, password, host, port):
    session = f"s-{uuid.uuid4().hex[:8]}"
    proxy = f"http://{base_user}-session-{session}:{password}@{host}:{port}"
    return httpx.Client(proxy=proxy, timeout=30, follow_redirects=True)

with flow_client("user", "pass", "gateway.example", 8000) as c:
    c.get("https://example.com/login")
    c.post("https://example.com/login", data={"u": "...", "p": "..."})
    r = c.get("https://example.com/account")

The client keeps cookies automatically, and the session identifier keeps the address. Both are required — see rotating vs sticky sessions.

#Measuring where time actually goes

“Slow” is not a diagnosis. Separate connection setup from the response:

import time, httpx

def timed_get(client, url):
    t0 = time.perf_counter()
    r = client.get(url)
    total = time.perf_counter() - t0
    return {
        "status": r.status_code,
        "bytes": len(r.content),
        "total_s": round(total, 3),
    }

with httpx.Client(proxy=PROXY, timeout=30) as c:
    c.get("https://api.ipify.org")          # warm the connection first
    print(timed_get(c, "https://target.example/page"))

Warming the connection before measuring is what separates a real latency figure from a one-off handshake cost. Measuring the first request of a fresh client measures the handshake, not the target.

#Verifying your exit before you rely on it

import httpx

with httpx.Client(proxy=PROXY, timeout=20) as c:
    me = c.get("http://ip-api.com/json/?fields=status,query,country,city,isp,as").json()
    assert me.get("status") == "success", me
    print(me["query"], me["country"], me.get("isp"))

Run this at the start of a job and log the result. When output looks wrong three hours in, knowing which country you were actually exiting from turns a mystery into a fact.

#Concurrency without causing your own outage

A provider’s concurrency cap is a maximum, not a target. Exceeding it behaves differently per provider — some queue, some refuse, some return an error indistinguishable from a target failure. Set your own ceiling below theirs:

import asyncio, httpx

async def run(urls, workers=8):
    q = asyncio.Queue()
    for u in urls:
        q.put_nowait(u)
    results = []

    async with httpx.AsyncClient(proxy=PROXY, timeout=30) as client:
        async def worker():
            while not q.empty():
                url = await q.get()
                try:
                    r = await client.get(url)
                    results.append((url, r.status_code))
                except httpx.HTTPError as e:
                    results.append((url, repr(e)))
                finally:
                    q.task_done()
                await asyncio.sleep(0.15)      # pace, do not sprint

        await asyncio.gather(*(worker() for _ in range(workers)))
    return results

The small sleep is deliberate. Hammering one destination from many addresses simultaneously is itself a detectable pattern, and a reliable way to earn 429s that no amount of rotation will fix.

#Common mistakes, in order of cost

#Putting it together: a small, well-behaved fetcher

The pieces above combine into something short enough to read in one sitting and robust enough to run unattended:

import asyncio, random, httpx

PROXY = "http://user:[email protected]:8000"
BLOCK_MARKERS = ("captcha", "unusual traffic", "access denied")

def usable(r):
    if r.status_code != 200 or len(r.content) < 1_500:
        return False
    low = r.text.lower()
    return not any(m in low for m in BLOCK_MARKERS)

async def fetch(client, url, attempts=4):
    for i in range(attempts):
        try:
            r = await client.get(url)
        except httpx.HTTPError:
            await asyncio.sleep((2 ** i) * (0.5 + random.random() * 0.5))
            continue

        if r.status_code == 429:
            wait = float(r.headers.get("Retry-After", 2 ** i))
            await asyncio.sleep(wait)
            continue
        if usable(r):
            return r
        if r.status_code in (403, 407):
            return None            # not worth retrying identically
        await asyncio.sleep((2 ** i) * (0.5 + random.random() * 0.5))
    return None

async def main(urls, workers=8):
    limits = httpx.Limits(max_connections=workers, max_keepalive_connections=workers)
    async with httpx.AsyncClient(proxy=PROXY, timeout=httpx.Timeout(30, connect=5),
                                 limits=limits, follow_redirects=True) as client:
        sem = asyncio.Semaphore(workers)
        async def one(u):
            async with sem:
                r = await fetch(client, u)
                await asyncio.sleep(0.1)
                return u, (r.status_code if r else None)
        return await asyncio.gather(*(one(u) for u in urls))

Every decision in that function is one of the failure modes described earlier. It reuses the client so handshakes are not repeated. It honours Retry-After. It applies jitter so workers do not resynchronise. It gives up immediately on 407, because retrying bad credentials cannot succeed. And it validates content rather than trusting the status code, which is the difference between a dataset and a directory full of block pages.

#Instrumenting it

Log four fields per request and most future incidents answer themselves: the status, the response size, the exit address in use, and the elapsed time. Size alongside status catches silent blocks; exit address alongside failure tells you whether one part of the pool is degraded rather than the whole thing.

#Choosing between the three libraries

All three will drive a proxy correctly, so the decision rests on what else you need.

A reasonable default is httpx: it covers both synchronous and asynchronous code with one API, handles SOCKS without an extra dependency, and supports HTTP/2 where the target offers it. requests remains the pragmatic choice for simple synchronous scripts and for the weight of existing examples. aiohttp suits codebases already committed to it, though passing the proxy per request rather than per session is easy to get subtly wrong across a large codebase.

Whichever you pick, the constraints described above are unchanged by the choice. None of them alters your TLS fingerprint, none removes the need to validate response content, and none makes an uncapped retry loop safe.