Python httpx and aiohttp with AeonProxy proxies

Protocols

HTTP and HTTPS sites over an HTTP proxy, or SOCKS5 with httpx[socks] or aiohttp-socks

Authentication

Login and password in the proxy URL

Rotation

A new IP per connection, or one IP per -sid- key for up to 60 minutes

Works with

Residential and mobile proxies

Before you start

• Python 3.10 or newer, then pip install httpx or pip install aiohttp, or both.
• Your proxy login and password from the Residential or Mobile page of the dashboard.
• Traffic on the product you use: a login without traffic gets 402.

httpx: rotating proxy

Pass the proxy URL with the login and password as proxy=. The scheme stays http:// for HTTPS sites: httpx opens a CONNECT tunnel through the proxy.

import httpx

proxy = "http://LOGIN-cc-us:[email protected]:7000"
limits = httpx.Limits(max_keepalive_connections=0)  # a new connection, so a new IP, per request

with httpx.Client(proxy=proxy, limits=limits, timeout=30) as client:
    for _ in range(3):
        print(client.get("https://api.ipify.org").text)

A Client sends the next request down the open connection, and the gateway picks the IP per connection: without limits, our three test requests shared one tunnel and one IP.

httpx: sticky session with AsyncClient

Add -sid- with a key of letters and digits, and -ttl- with the minutes to keep the IP (1–60). Every connection with the same key leaves from the same IP, so the five parallel requests below share it.

import asyncio
import secrets
import httpx

async def main():
    sid = secrets.token_hex(4)  # one key = one IP for up to ttl minutes
    proxy = f"http://LOGIN-cc-de-sid-{sid}-ttl-30:[email protected]:7000"
    async with httpx.AsyncClient(proxy=proxy, timeout=30) as client:
        replies = await asyncio.gather(*(client.get("https://api.ipify.org") for _ in range(5)))
        print([r.text for r in replies])  # the same IP five times

asyncio.run(main())

httpx: SOCKS5

Install pip install "httpx[socks]" and use port 7001. With socks5:// or socks5h://, httpx sends the host name to the proxy, so DNS is resolved there and not on your machine.

import httpx

proxy = "socks5://LOGIN-cc-gb:[email protected]:7001"
with httpx.Client(proxy=proxy, timeout=30) as client:
    print(client.get("https://api.ipify.org").text)

aiohttp: HTTP proxy

aiohttp takes proxy= per request or once per ClientSession. A session also reuses connections, so the connector below has force_close=True for a new IP per request.

import asyncio
import aiohttp

PROXY = "http://LOGIN-cc-us:[email protected]:7000"

async def main():
    connector = aiohttp.TCPConnector(force_close=True)  # no keep-alive: a new IP per request
    timeout = aiohttp.ClientTimeout(total=30)
    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 r:
                print(await r.text())

asyncio.run(main())

aiohttp: one sticky IP per worker

Give each task its own key. The three workers below run in parallel, each on its own IP for up to 15 minutes.

import asyncio
import secrets
import aiohttp

async def worker(session, n):
    sid = secrets.token_hex(4)  # each worker keeps its own IP
    proxy = f"http://LOGIN-cc-us-sid-{sid}-ttl-15:[email protected]:7000"
    for _ in range(2):
        async with session.get("https://api.ipify.org", proxy=proxy) as r:
            print(n, sid, await r.text())

async def main():
    async with aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=30)) as session:
        await asyncio.gather(*(worker(session, n) for n in range(3)))

asyncio.run(main())

aiohttp: SOCKS5

aiohttp has no SOCKS support of its own: a socks5:// URL in proxy= fails with Bad status line. Install aiohttp-socks and pass its connector to the session; rdns=True, its default, sends the host name to the proxy.

import asyncio
import aiohttp
from aiohttp_socks import ProxyConnector

async def main():
    connector = ProxyConnector.from_url("socks5://LOGIN-cc-gb:[email protected]:7001", rdns=True)
    async with aiohttp.ClientSession(connector=connector) as session:
        async with session.get("https://api.ipify.org") as r:
            print(await r.text())

asyncio.run(main())

If it does not work