Прокси в HTTPX: Client и AsyncClient

Короткие ответы

Проверьте версию и окружение Python

Примеры рассчитаны на Python 3.11 или новее и HTTPX 0.28.1. Установите библиотеку в окружение проекта и проверьте импорт тем интерпретатором, который запускает задачу. Терминал, notebook и фоновый worker могут использовать разные окружения. Не меняйте зависимости работающего приложения до проверки его lock-файла и совместимости остальных пакетов.

python -m pip install "httpx==0.28.1"
python -c "import httpx; print(httpx.__version__)"

Изменение аргумента описано в заметках о выпуске HTTPX 0.28. Здесь показана настройка одного HTTP-шлюза. Для другого клиента есть отдельные инструкции: Python Requests и aiohttp. Их параметры авторизации и управления соединениями не взаимозаменяемы.

Подготовьте адрес и авторизацию

Получите полное имя пользователя в личном кабинете, выбрав доступные пул и страну. Передайте это имя через переменную окружения PROXY_USER. Пароль прокси вводится отдельно через запрос без отображения ввода. Пароль аккаунта и ключ API здесь не подходят. Сохраните следующие три Python-файла в одной папке, чтобы обе проверки импортировали общий файл настроек.

import getpass
import os
import ssl
from urllib.parse import urlsplit

import certifi
import httpx


def connection_settings():
    server = os.environ.get("PROXY_SERVER", "http://gw.portproof.org:7000")
    address = urlsplit(server)
    if (address.scheme != "http" or not address.hostname
            or address.port is None or address.username is not None
            or address.password is not None or address.path not in ("", "/")
            or address.query or address.fragment):
        raise SystemExit("Use an HTTP gateway URL without credentials or a path")
    proxy = httpx.Proxy(
        server,
        auth=(os.environ["PROXY_USER"], getpass.getpass("Proxy password: ")),
    )
    tls = ssl.create_default_context(
        cafile=os.environ.get("PROXY_CA_BUNDLE") or certifi.where()
    )
    return {
        "proxy": proxy,
        "verify": tls,
        "trust_env": False,
        "follow_redirects": False,
        "timeout": httpx.Timeout(10.0, connect=5.0, pool=2.0),
    }

Имя и пароль передаются парой auth в объект Proxy. Поэтому знаки пунктуации в пароле не превращаются в части URL. Это авторизация на шлюзе: не переносите её в auth конечного запроса или обычный заголовок Authorization. Не выводите объект настроек в журнал. В фоновой задаче замените интерактивный ввод обращением к своему хранилищу секретов до начала сетевой работы.

Параметр trust_env=False исключает влияние HTTP_PROXY, HTTPS_PROXY и NO_PROXY на выбранный маршрут. Проверка сертификатов также задана явно: используется certifi либо одобренный вами набор PROXY_CA_BUNDLE. Переменная SSL_CERT_FILE в этом примере не выбирает набор сертификатов. Если в вашей сети обязателен корпоративный шлюз или сертификат, настройте их осознанно, сохранив требования сети.

Сделайте один запрос через Client

import json

import httpx

from proxy_settings import connection_settings

settings = connection_settings()
try:
    with httpx.Client(**settings) as client:
        with client.stream("GET", "https://api.portproof.org/v1/echo-ip") as response:
            response.raise_for_status()
            body = bytearray()
            for chunk in response.iter_bytes(chunk_size=8192):
                if len(body) + len(chunk) > 65536:
                    raise ValueError("Echo response exceeds 64 KiB")
                body.extend(chunk)
            payload = json.loads(body)
            if not isinstance(payload, dict) or not isinstance(payload.get("ip"), str):
                raise ValueError("Expected an IP address in the echo response")
            print("HTTP", response.status_code, "exit", payload["ip"])
except (httpx.HTTPError, ValueError) as error:
    raise SystemExit(f"Check failed: {type(error).__name__}") from None

Запустите python httpx_check.py. Скрипт выводит HTTP-код и значение поля ip, полученное от echo-сервиса. Это адрес, который увидел сервис в данном запросе; он не подтверждает страну выхода или доступность другого сайта. Сначала убедитесь, что ответ имеет ожидаемый формат, затем переходите к одной разрешённой операции вашего приложения.

В примере отключены переходы по перенаправлениям и нет автоматических повторов. Тело читается частями; накопление прекращается, если декодированный ответ превышает 64 КиБ. Этот предел относится к памяти диагностического скрипта, а не к оплачиваемым сетевым байтам. Контекстные менеджеры закрывают ответ и клиент, в том числе после ошибки разбора JSON.

Используйте AsyncClient в асинхронной задаче

import asyncio
import json

import httpx

from proxy_settings import connection_settings

settings = connection_settings()


async def main():
    try:
        async with asyncio.timeout(20):
            async with httpx.AsyncClient(**settings) as client:
                async with client.stream("GET", "https://api.portproof.org/v1/echo-ip") as response:
                    response.raise_for_status()
                    body = bytearray()
                    async for chunk in response.aiter_bytes(chunk_size=8192):
                        if len(body) + len(chunk) > 65536:
                            raise ValueError("Echo response exceeds 64 KiB")
                        body.extend(chunk)
                    payload = json.loads(body)
                    if not isinstance(payload, dict) or not isinstance(payload.get("ip"), str):
                        raise ValueError("Expected an IP address in the echo response")
                    print("HTTP", response.status_code, "exit", payload["ip"])
    except (httpx.HTTPError, ValueError, TimeoutError) as error:
        raise SystemExit(f"Check failed: {type(error).__name__}") from None


if __name__ == "__main__":
    asyncio.run(main())

Запустите python httpx_async_check.py как отдельный процесс. Если цикл событий уже работает, вызывайте await main(), а не вложенный asyncio.run(). Настройки и пароль должны быть получены заранее. Ожидание человека у скрытого приглашения не входит в сетевой таймаут и не подходит для запуска по расписанию без оператора.

HTTPX отдельно ограничивает ожидание соединения, чтения, записи и свободного соединения в пуле. Таймаут чтения ограничивает ожидание очередных данных: медленная передача может длиться дольше этого значения. В асинхронном варианте asyncio.timeout дополнительно запрашивает отмену через двадцать секунд. Освобождение ресурсов может занять дополнительное время. Для синхронной задачи задайте общий срок выполнения в worker или планировщике.

Создавайте клиент на срок связанной работы и переиспользуйте его, закрывая после завершения. Клиент внутри каждой итерации создаёт лишние соединения. При параллельной обработке отдельно ограничьте очередь задач: размер пула соединений не заменяет предел объёма работы. Подробнее: таймауты HTTPX и асинхронные клиенты.

Что подтверждено локальной проверкой

Эти же примеры выполнялись 30 сентября 2026 года с Python 3.12.14, HTTPX 0.28.1 и httpcore 1.0.9. Локальный HTTP-прокси с тестовыми учётными данными создавал CONNECT-туннели к локальному HTTPS-серверу. Русская версия использует тот же исполняемый код. Проверка не обращалась к рабочему пулу и не измеряла его скорость.

• Client и AsyncClient успешно авторизовались с пунктуацией в пароле; сервер назначения не получил пароль прокси или заголовок Authorization.
• Конфликтующие настройки прокси в окружении не меняли явно выбранный маршрут.
• Неверный пароль приводил к ProxyError после ответа 407, а недоверенный сертификат — к ConnectError.
• Задержка чтения и общий асинхронный предел завершались соответствующими исключениями; закрытие ответа и клиента проверялось.

Отделите ошибку соединения от ответа сайта

ProxyError требует проверки шлюза и авторизации. Начните с разбора 407 либо ошибок CONNECT — на английском. HTTPStatusError после raise_for_status() относится к полученному HTTP-ответу; выясните, какой сервис его вернул. При ошибке сертификата проверяйте имя сайта, системное время и цепочку доверия, сохраняя проверку TLS.

Переиспользование соединения не означает смену выходного адреса при каждом вызове HTTPX. Параметры ротации и sticky-сессии задаются в имени пользователя шлюза. Sticky удерживает устройство, но его адрес может измениться. Сверяйте результат конкретного запроса; подробнее в руководстве по сессиям.

После подключения измерьте небольшой типичный запуск: полезные ответы, попытки, длительность и расход в кабинете. Повторять операции с побочными эффектами автоматически нельзя. Для выбора объёма используйте расчёт трафика — на английском, а для покупки сверяйте актуальные тарифы — на английском. Оба общих пула используют баланс в ГБ.