How to use a proxy with aiohttp

pip install aiohttppip install aiohttp-socks   # optional, for SOCKS5

import asyncioimport aiohttpPROXY = "http://USERNAME_pool-arcfast_country-de:PASSWORD@proxy.arcproxies.io:8000"async def main():    timeout = aiohttp.ClientTimeout(total=30)    async with aiohttp.ClientSession(timeout=timeout) as session:        async with session.get("https://api.ipify.org?format=json", proxy=PROXY) as r:            print(await r.json())   # {"ip": "…"}asyncio.run(main())

import asyncioimport secretsimport aiohttpasync def crawl_listing(session, start_url, pages=3):    proxy = f"http://USERNAME_pool-arcfast_country-de_session-{secrets.token_hex(4)}_sesstime-10:PASSWORD@proxy.arcproxies.io:8000"    statuses = []    for page in range(1, pages + 1):        async with session.get(start_url, params={"page": page}, proxy=proxy) as r:            await r.read()            statuses.append(r.status)    return statusesasync def main(starts):    connector = aiohttp.TCPConnector(limit=50)    limit = asyncio.Semaphore(10)    async with aiohttp.ClientSession(connector=connector,                                     timeout=aiohttp.ClientTimeout(total=30)) as session:        async def guarded(url):            async with limit:                return await crawl_listing(session, url)        print(await asyncio.gather(*(guarded(u) for u in starts)))asyncio.run(main(["https://example.com/c/lamps", "https://example.com/c/chairs"]))

MARKETS = {    "pl-warsaw": "USERNAME_pool-arcfast_country-pl_city-warsaw",    "it-milan":  "USERNAME_pool-arcfast_country-it_city-milan",    "de":        "USERNAME_pool-arcfast_country-de",}async def prices(session, url):    out = {}    for market, user in MARKETS.items():        async with session.get(url, proxy=f"http://{user}:PASSWORD@proxy.arcproxies.io:8000") as r:            out[market] = r.status    return out

import aiohttpfrom aiohttp_socks import ProxyConnectorasync def main():    connector = ProxyConnector.from_url("socks5://USERNAME_pool-arcfast_country-de:PASSWORD@proxy.arcproxies.io:8000")    async with aiohttp.ClientSession(connector=connector) as session:        async with session.get("https://api.ipify.org?format=json") as r:            print(await r.json())

import asyncioimport secretsimport aiohttpasync def fetch(session, url, attempts=4):    for attempt in range(attempts):        proxy = f"http://USERNAME_pool-arcfast_country-de_session-{secrets.token_hex(4)}_sesstime-5:PASSWORD@proxy.arcproxies.io:8000"        try:            async with session.get(url, proxy=proxy) as r:                if r.status == 407:                    raise RuntimeError("proxy rejected the credentials")                if r.status != 429 and r.status < 500:                    return await r.text()        except aiohttp.ClientHttpProxyError as err:            if err.status == 407:                raise RuntimeError("proxy rejected the credentials") from None            print(f"proxy error {err.status}")       # not str(err): it contains the URL        except (aiohttp.ClientConnectionError, asyncio.TimeoutError):            pass        await asyncio.sleep(2 ** attempt)    raise RuntimeError(f"gave up on {url}")