За една асинхронна заявка през 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 само когато наследената среда на процеса е желаната политика за маршрутизиране.

Подобни статии