Використання проксі в Python з httpx

Використання проксі в Python за допомогою httpx

httpx та проксі

httpx — це сучасна HTTP-бібліотека для Python, яка підтримує як синхронний, так і асинхронний режими роботи. Ключовою перевагою є вбудована підтримка проксі HTTP/2 та SOCKS без додаткових бібліотек.

Встановлення

pip install httpx
# Для підтримки SOCKS:
pip install httpx[socks]
# Для HTTP/2:
pip install httpx[http2]

Синхронний режим

HTTP Проксі

import httpx

proxy = "http://user:pass@proxy_ip:port"
response = httpx.get("https://httpbin.org/ip", proxy=proxy)
print(response.json())

Різні проксі для HTTP та HTTPS

proxies = {
    "http://": "http://proxy1:port",
    "https://": "http://proxy2:port",
}

with httpx.Client(proxy=proxies) as client:
    response = client.get("https://httpbin.org/ip")
    print(response.json())

SOCKS5 Проксі

import httpx

proxy = "socks5://user:pass@proxy_ip:port"
with httpx.Client(proxy=proxy) as client:
    response = client.get("https://httpbin.org/ip")
    print(response.json())

Асинхронний режим

Базовий приклад

import httpx
import asyncio

async def fetch():
    proxy = "http://user:pass@proxy_ip:port"
    async with httpx.AsyncClient(proxy=proxy) as client:
        response = await client.get("https://httpbin.org/ip")
        print(response.json())

asyncio.run(fetch())

Паралельні запити

import httpx
import asyncio

async def fetch_url(client, url):
    try:
        response = await client.get(url, timeout=10)
        return response.text
    except Exception as e:
        return None

async def main():
    proxy = "http://user:pass@proxy_ip:port"
    urls = [f"https://example.com/page/{i}" for i in range(50)]

    async with httpx.AsyncClient(proxy=proxy) as client:
        tasks = [fetch_url(client, url) for url in urls]
        results = await asyncio.gather(*tasks)
        success = sum(1 for r in results if r)
        print(f"Success: {success}/{len(urls)}")

asyncio.run(main())

HTTP/2 через проксі

httpx — одна з небагатьох бібліотек Python з підтримкою HTTP/2:

import httpx

proxy = "http://user:pass@proxy_ip:port"
with httpx.Client(proxy=proxy, http2=True) as client:
    response = client.get("https://httpbin.org/ip")
    print(f"HTTP version: {response.http_version}")
    print(response.json())

HTTP/2 через проксі працює за допомогою методу CONNECT — проксі створює TCP-тунель, через який встановлюється HTTP/2-з'єднання з сервером.

Ротація проксі

import httpx
import random

PROXIES = [
    "http://user:pass@proxy1:port",
    "http://user:pass@proxy2:port",
    "http://user:pass@proxy3:port",
]

def get_random_proxy():
    return random.choice(PROXIES)

# Синхронна ротація
for i in range(10):
    proxy = get_random_proxy()
    with httpx.Client(proxy=proxy) as client:
        resp = client.get("https://httpbin.org/ip")
        print(resp.json())

Асинхронна ротація

import httpx
import asyncio
import random

PROXIES = [
    "http://user:pass@proxy1:port",
    "http://user:pass@proxy2:port",
    "http://user:pass@proxy3:port",
]

async def fetch_with_rotation(url):
    proxy = random.choice(PROXIES)
    async with httpx.AsyncClient(proxy=proxy) as client:
        try:
            resp = await client.get(url, timeout=10)
            return resp.text
        except Exception:
            return None

async def main():
    urls = ["https://example.com"] * 20
    tasks = [fetch_with_rotation(url) for url in urls]
    results = await asyncio.gather(*tasks)
    print(f"Success: {sum(1 for r in results if r)}")

asyncio.run(main())

Конфігурація клієнта

Таймаут

timeout = httpx.Timeout(
    connect=5.0,    # з'єднання з проксі та сервером
    read=10.0,      # читання відповіді
    write=5.0,      # надсилання запиту
    pool=5.0        # очікування в пулі з'єднань
)

client = httpx.Client(proxy=proxy, timeout=timeout)

Заголовки та Cookies

headers = {
    "User-Agent": "Mozilla/5.0 ...",
    "Accept-Language": "en-US,en;q=0.9",
}

client = httpx.Client(
    proxy=proxy,
    headers=headers,
    follow_redirects=True,
    max_redirects=10
)

Пул з'єднань

limits = httpx.Limits(
    max_connections=100,          # загальна кількість з'єднань
    max_keepalive_connections=20  # постійні з'єднання (keep-alive)
)

client = httpx.AsyncClient(proxy=proxy, limits=limits)

Обробка помилок

import httpx

async def safe_fetch(client, url):
    try:
        response = await client.get(url)
        response.raise_for_status()
        return response.text
    except httpx.ProxyError as e:
        print(f"Помилка проксі: {e}")
    except httpx.ConnectTimeout:
        print("Таймаут з'єднання")
    except httpx.ReadTimeout:
        print("Таймаут читання")
    except httpx.HTTPStatusError as e:
        print(f"HTTP {e.response.status_code}")
    except httpx.RequestError as e:
        print(f"Помилка запиту: {e}")
    return None

httpx проти requests проти aiohttp

Міграція з requests

httpx розроблений як пряма заміна для requests. Міграція мінімальна:

# requests
import requests
resp = requests.get(url, proxies={"https": proxy})

# httpx
import httpx
resp = httpx.get(url, proxy=proxy)

Основна відмінність: в httpx використовується параметр proxy (в однині) замість proxies.

Висновок

httpx — найкращий вибір для нових проєктів на Python, що потребують функціональності проксі. Його вбудована підтримка HTTP/2, SOCKS5 та як синхронного, так і асинхронного режимів робить його універсальним інструментом. Сумісний з requests API спрощує міграцію для існуючих проєктів.