Адзін запыт aiohttp праз HTTP-проксі

Для аднаго асінхроннага запыту праз HTTP-проксі без аўтэнтыфікацыі задайце маршрут на session.get(). Пры патрэбе дадайце aiohttp ў праект:

uv add aiohttp

Замяніце дакументацыйны адрас ніжэй сваім HTTP-слухачом або кандыдатам са спіса HTTP-проксі. Ён павінен дазваляць CONNECT да HTTPS-мэты; пункт толькі з SOCKS патрабуе адпаведнага канектара замест гэтага аргумента proxy.

import asyncio
import aiohttp

async def main():
    timeout = aiohttp.ClientTimeout(total=15, connect=5)
    async with aiohttp.ClientSession(
        timeout=timeout,
        trust_env=False,
    ) as session:
        async with session.get(
            "https://example.com/",
            proxy="http://203.0.113.10:8080",
        ) as response:
            print(response.status)
            print((await response.text())[:200])

asyncio.run(main())

Запыт выводзіць статус HTTP і першыя 200 сімвалаў тэксту адказу. Адказ 403 або 429 тут застаецца даступным для праверкі; выклічце response.raise_for_status(), калі праграма павінна ператвараць няўдалыя статусы ў выключэнні. Асінхронныя менеджары кантэксту закрываюць сесію і адказ пасля завяршэння блокаў, у тым ліку пры выключэнні.

15-секундны бюджэт total ахоплівае аперацыю запыту. 5-секундны бюджэт connect уключае чаканне даступнага злучэння з пула, таму заняты пул можа вычарпаць яго да адкрыцця новага сокета. Гэта важна пры пераўтварэнні аднаго выкліку ў адначасовыя праверкі.

trust_env=False захоўвае паказаны маршрут яўным. aiohttp па змаўчанні не знаходзіць зменныя асяроддзя проксі; экспарт такой зменнай у іншым тэрмінале не наладзіць гэты выклік. Выбірайце trust_env=True толькі тады, калі ўспадкаванае асяроддзе працэсу — патрэбная палітыка маршрутызацыі.