Python'da httpx ile proxy kullanımı

Python'da httpx ile proxy kullanımı

httpx ve proxy'ler

httpx, Python için hem senkron hem de asenkron çalışma modlarını destekleyen modern bir HTTP kütüphanesidir. Temel avantajı, ek kütüphane gerektirmeden HTTP/2 ve SOCKS proxy'lere yerleşik destek sunmasıdır.

Kurulum

pip install httpx
# SOCKS desteği için:
pip install httpx[socks]
# HTTP/2 için:
pip install httpx[http2]

Senkron mod

HTTP proxy

import httpx

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

HTTP ve HTTPS için farklı proxy'ler

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 proxy

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())

Asenkron mod

Temel örnek

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())

Paralel istekler

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())

Proxy üzerinden HTTP/2

httpx, HTTP/2 destekleyen az sayıdaki Python kütüphanesinden biridir:

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())

Proxy üzerinden HTTP/2, CONNECT yöntemiyle çalışır — proxy bir TCP tüneli oluşturur ve sunucuya HTTP/2 bağlantısı bu tünel üzerinden kurulur.

Proxy rotasyonu

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)

# Senkron rotasyon
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())

Asenkron rotasyon

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())

İstemci yapılandırması

Timeout

timeout = httpx.Timeout(
    connect=5.0,    # proxy ve sunucuya bağlantı
    read=10.0,      # yanıtın okunması
    write=5.0,      # isteğin gönderilmesi
    pool=5.0        # bağlantı havuzunda bekleme
)

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

Başlıklar ve çerezler

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
)

Bağlantı havuzu

limits = httpx.Limits(
    max_connections=100,          # toplam bağlantı
    max_keepalive_connections=20  # keep-alive bağlantıları
)

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

Hata yönetimi

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"Proxy error: {e}")
    except httpx.ConnectTimeout:
        print("Connection timeout")
    except httpx.ReadTimeout:
        print("Read timeout")
    except httpx.HTTPStatusError as e:
        print(f"HTTP {e.response.status_code}")
    except httpx.RequestError as e:
        print(f"Request error: {e}")
    return None

httpx vs requests vs aiohttp

requests'ten geçiş

httpx, requests'in doğrudan yerine geçecek şekilde tasarlanmıştır. Geçiş çok azdır:

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

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

Temel fark: httpx'te proxies yerine proxy parametresi (tekil) kullanılır.

Sonuç

httpx, proxy işlevselliği gerektiren yeni Python projeleri için en iyi seçimdir. HTTP/2, SOCKS5 ve hem senkron hem asenkron modlara yerleşik desteği onu çok yönlü bir araç yapar. requests uyumlu API'si, mevcut projelerin geçişini kolaylaştırır.