Налаштування ротації проксі в Python: requests, aiohttp, scrapy

Робочі приклади коду на Python для ротації проксі у популярних бібліотеках. Custom retry-logic, proxy pool management, error handling, rate limiting.

requests: базова ротація

Найпростіший варіант для скриптів on-demand. Передаєте проксі у виклик:

import requests
import random

PROXIES = [
    'http://user:[email protected]:7777',
    'http://user:[email protected]:7777',
    'http://user:[email protected]:7777',
]

def fetch(url):
    proxy = random.choice(PROXIES)
    response = requests.get(
        url,
        proxies={'http': proxy, 'https': proxy},
        timeout=15
    )
    return response

# Usage
data = fetch('https://api.ipify.org')
print(data.text)  # Кожен запит з нового IP

Для residential з sticky-сесією — передавайте session-token у user:

import secrets
session_id = secrets.token_hex(8)
proxy = f'http://customer-XXX-session-{session_id}:[email protected]:7000'
# Тепер всі запити з цим session-token йдуть з того самого IP

requests: proxy pool з retry і блек-листом

Production-ready клас з обробкою failed proxies:

import requests
import time
from collections import deque
from threading import Lock

class ProxyPool:
    def __init__(self, proxies, max_failures=3, blacklist_ttl=300):
        self.proxies = deque(proxies)
        self.failures = {}
        self.blacklist = {}
        self.max_failures = max_failures
        self.blacklist_ttl = blacklist_ttl
        self.lock = Lock()

    def get(self):
        with self.lock:
            now = time.time()
            # Очищуємо blacklist
            self.blacklist = {p: t for p, t in self.blacklist.items()
                              if now - t < self.blacklist_ttl}
            # Шукаємо вільний
            for _ in range(len(self.proxies)):
                proxy = self.proxies[0]
                self.proxies.rotate(-1)
                if proxy not in self.blacklist:
                    return proxy
            raise RuntimeError('All proxies blacklisted')

    def report_failure(self, proxy):
        with self.lock:
            self.failures[proxy] = self.failures.get(proxy, 0) + 1
            if self.failures[proxy] >= self.max_failures:
                self.blacklist[proxy] = time.time()
                self.failures[proxy] = 0

    def report_success(self, proxy):
        with self.lock:
            self.failures.pop(proxy, None)

def fetch_with_retry(url, pool, max_retries=5):
    for attempt in range(max_retries):
        proxy = pool.get()
        try:
            r = requests.get(
                url,
                proxies={'http': proxy, 'https': proxy},
                timeout=15
            )
            if r.status_code in (200, 304):
                pool.report_success(proxy)
                return r
            elif r.status_code in (429, 503):
                pool.report_failure(proxy)
                continue
        except (requests.Timeout, requests.ConnectionError):
            pool.report_failure(proxy)
    return None

aiohttp: async ротація для високого throughput

Для парсингу 10K+ URL/хвилину requests не вистачає. aiohttp дає async-throughput:

import asyncio
import aiohttp
import random

PROXIES = ['http://user:[email protected]:7777', ...]
SEMAPHORE = asyncio.Semaphore(50)  # 50 паралельних з'єднань

async def fetch(session, url):
    async with SEMAPHORE:
        proxy = random.choice(PROXIES)
        try:
            async with session.get(
                url,
                proxy=proxy,
                timeout=aiohttp.ClientTimeout(total=15)
            ) as r:
                if r.status == 200:
                    return await r.text()
                return None
        except (asyncio.TimeoutError, aiohttp.ClientError):
            return None

async def fetch_all(urls):
    async with aiohttp.ClientSession() as session:
        tasks = [fetch(session, u) for u in urls]
        return await asyncio.gather(*tasks)

# Usage
urls = ['https://example.com/' + str(i) for i in range(1000)]
results = asyncio.run(fetch_all(urls))

Scrapy: ProxyMiddleware з ротацією

Scrapy має built-in proxy support, але для ротації потрібно custom middleware:

# middlewares.py
import random

class RotatingProxyMiddleware:
    def __init__(self, proxies):
        self.proxies = proxies

    @classmethod
    def from_crawler(cls, crawler):
        return cls(crawler.settings.getlist('PROXY_POOL'))

    def process_request(self, request, spider):
        request.meta['proxy'] = random.choice(self.proxies)

    def process_exception(self, request, exception, spider):
        # Якщо проксі вмер — повторюємо з іншим
        request.meta['proxy'] = random.choice(self.proxies)
        return request

# settings.py
DOWNLOADER_MIDDLEWARES = {
    'myproject.middlewares.RotatingProxyMiddleware': 350,
}
PROXY_POOL = [
    'http://user:[email protected]:7777',
    'http://user:[email protected]:7777',
]
DOWNLOAD_TIMEOUT = 15
RETRY_TIMES = 3

Типові проблеми і fixes

• SSL errors з residential проксі — додайте verify=False для debug. У production налаштуйте свій CA-bundle.
• HTTP 407 (Proxy Authentication) — перевірте формат user:password. Деякі провайдери вимагають URL-encode спецсимволи.
• Connection reset на residential — нормально, проксі-сесії виходять з строю. Реалізуйте retry-logic.
• HTTP 429 від цільового сайту — rate-limit. Додайте sleep і ротуйте IP агресивніше.
• "Tunnel connection failed" — формат user:pass неправильний для HTTPS. Спробуйте https:// в URL проксі.

Для серйозних проектів використовуйте окремий пакет типу requests-ip-rotator (для AWS API Gateway as proxy) або enterprise-рішення Bright Data SDK.