Rotating residential proxies in Python: requests, httpx and aiohttp

import os

import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

USER = os.environ.get("CRAWLRELAY_USER", "USERNAME-country-us")
PASSWORD = os.environ.get("CRAWLRELAY_PASSWORD", "PASSWORD")
PROXY = "http://%s:%[email protected]:7777" % (USER, PASSWORD)

retry = Retry(
    total=4,
    connect=2,
    backoff_factor=1.0,  # waits of 0 s, 2 s, 4 s, 8 s (Retry-After wins on 429/503)
    status_forcelist=(429, 502, 503, 504),
    allowed_methods=frozenset({"GET", "HEAD"}),
    respect_retry_after_header=True,
)

session = requests.Session()
session.trust_env = False  # ignore HTTP(S)_PROXY from the environment
session.mount("http://", HTTPAdapter(max_retries=retry))
session.mount("https://", HTTPAdapter(max_retries=retry))
session.proxies = {"http": PROXY, "https": PROXY}
session.headers["User-Agent"] = "ExampleCrawler/1.0 (+https://example.com/crawler-info)"
session.headers["Connection"] = "close"  # new connection per request, so usually a new exit IP

r = session.get("https://api.ipify.org", timeout=(10, 30))
print(r.status_code, r.text)

import re
from urllib.parse import quote

ORDER = ("country", "state", "city", "zip", "asn", "session", "ttl")
VALUE = re.compile(r"^[a-z0-9_]+$")


def build_username(base, **params):
    """build_username("USERNAME", country="us", city="new_york", session="job1", ttl=30)"""
    parts = [base]
    for key in ORDER:
        value = params.pop(key, None)
        if value is None:
            continue
        value = str(value).lower()
        if not VALUE.match(value):
            raise ValueError("bad value for %s: %r" % (key, value))
        if key == "ttl" and not 1 <= int(value) <= 30:
            raise ValueError("ttl must be 1 to 30 minutes")
        parts.append("%s-%s" % (key, value))
    if params:
        raise ValueError("unknown parameters: %s" % ", ".join(sorted(params)))
    return "-".join(parts)


def proxy_url(username, password, socks=False):
    scheme, port = ("socks5h", 7778) if socks else ("http", 7777)
    return "%s://%s:%[email protected]:%d" % (
        scheme, quote(username, safe=""), quote(password, safe=""), port)


print(proxy_url(build_username("USERNAME", country="us", state="ny", city="new_york", session="job1", ttl=20), "p@ss:word"))

import httpx

PROXY = "http://USERNAME-country-gb:[email protected]:7777"

# No keep-alive: every request opens a new connection, so it usually gets a new exit IP
with httpx.Client(
    proxy=PROXY,
    timeout=httpx.Timeout(30.0, connect=10.0),
    limits=httpx.Limits(max_keepalive_connections=0),
    follow_redirects=True,
) as client:
    for _ in range(3):
        r = client.get("https://api.ipify.org")
        print(r.text, r.num_bytes_downloaded, "bytes on the wire")

import asyncio
from collections import defaultdict
from urllib.parse import urlsplit

import httpx

PROXY = "http://USERNAME-country-us:[email protected]:7777"
PER_HOST = 4  # parallel requests per target host
host_slots = defaultdict(lambda: asyncio.Semaphore(PER_HOST))


async def fetch(client, url):
    async with host_slots[urlsplit(url).hostname]:
        try:
            r = await client.get(url)
            return url, r.status_code, r.num_bytes_downloaded
        except httpx.HTTPError as exc:
            return url, type(exc).__name__, 0


async def main(urls):
    limits = httpx.Limits(max_connections=50, max_keepalive_connections=0)
    timeout = httpx.Timeout(30.0, connect=10.0)
    async with httpx.AsyncClient(proxy=PROXY, limits=limits, timeout=timeout) as client:
        for result in await asyncio.gather(*(fetch(client, u) for u in urls)):
            print(*result)


asyncio.run(main(["https://example.com/", "https://example.org/", "https://example.net/"]))

import asyncio

import aiohttp

# Credentials go in the proxy URL; percent-encode a password that contains @ or :
PROXY = "http://USERNAME-country-fr:[email protected]:7777"


async def main():
    # force_close=True: one connection per request, so each request usually gets a new exit IP
    connector = aiohttp.TCPConnector(limit=20, force_close=True)
    timeout = aiohttp.ClientTimeout(total=60, sock_connect=10)
    async with aiohttp.ClientSession(connector=connector, timeout=timeout) as session:
        for _ in range(3):
            async with session.get("https://api.ipify.org", proxy=PROXY) as resp:
                print(resp.status, await resp.text())


asyncio.run(main())

import uuid

import requests

session_id = uuid.uuid4().hex[:10]  # one ID per flow, lowercase letters and digits
user = "USERNAME-country-us-state-ca-session-%s-ttl-20" % session_id
proxy = "http://%s:[email protected]:7777" % user

with requests.Session() as s:
    s.proxies = {"http": proxy, "https": proxy}
    first = s.get("https://api.ipify.org", timeout=(10, 30)).text
    second = s.get("https://api.ipify.org", timeout=(10, 30)).text
    print(first, second, "same exit" if first == second else "exit changed")