Python aiohttp 异步代理接入示例

import asyncio
import os

import aiohttp


async def main() -> None:
    """通过显式 HTTP 代理异步读取公开出口检测结果。"""
    proxy_url = (
        f"http://{os.environ['PROXY_HOST']}"
        f":{os.environ['PROXY_PORT']}"
    )
    proxy_auth = aiohttp.BasicAuth(
        os.environ["PROXY_USERNAME"],
        os.environ["PROXY_PASSWORD"],
    )
    timeout = aiohttp.ClientTimeout(total=30, connect=10)

    async with aiohttp.ClientSession(timeout=timeout) as session:
        async with session.get(
            "https://<public-test-host>/ip",
            proxy=proxy_url,
            proxy_auth=proxy_auth,
        ) as response:
            response.raise_for_status()
            print(await response.json())


asyncio.run(main())