Прокси в Python requests и aiohttp: настройка за 5 минут

При разработке парсеров, автоматизации сбора данных или тестировании веб-сервисов из Python часто требуется использовать прокси-серверы. Библиотеки requests и aiohttp предоставляют гибкие механизмы работы с прокси, но их настройка имеет важные нюансы. В этом руководстве разберём синхронные и асинхронные подходы, покажем примеры для HTTP и SOCKS5 прокси, рассмотрим ротацию IP и обработку ошибок.

Базовая настройка прокси в requests

Библиотека requests — это стандарт для HTTP-запросов в Python. Настройка прокси выполняется через параметр proxies, который принимает словарь с протоколами и адресами прокси-серверов.

Простейший пример с HTTP-прокси:

import requests

# Настройка прокси
proxies = {
    'http': 'http://123.45.67.89:8080',
    'https': 'http://123.45.67.89:8080'
}

# Выполнение запроса через прокси
response = requests.get('https://httpbin.org/ip', proxies=proxies)
print(response.json())  # {'origin': '123.45.67.89'}

Обратите внимание: для HTTPS-запросов также указывается протокол http:// в значении прокси (не https://). Это связано с тем, что соединение с прокси-сервером устанавливается по HTTP, а затем через метод CONNECT создаётся туннель для HTTPS-трафика.

Использование переменных окружения:

Библиотека requests автоматически читает прокси из переменных окружения HTTP_PROXY и HTTPS_PROXY:

import os
import requests

# Установка через переменные окружения
os.environ['HTTP_PROXY'] = 'http://123.45.67.89:8080'
os.environ['HTTPS_PROXY'] = 'http://123.45.67.89:8080'

# Прокси применятся автоматически
response = requests.get('https://httpbin.org/ip')
print(response.json())

Этот подход удобен для контейнеризации (Docker) или когда прокси настраиваются на уровне системы. Однако для гибкости рекомендуется явная передача параметра proxies.

Аутентификация и SOCKS5 в requests

Большинство коммерческих прокси-сервисов требуют аутентификацию по логину и паролю. В requests это реализуется через URL-формат с учётными данными.

HTTP-прокси с аутентификацией:

import requests

# Формат: http://username:password@host:port
proxies = {
    'http': 'http://user123:pass456@proxy.example.com:8080',
    'https': 'http://user123:pass456@proxy.example.com:8080'
}

response = requests.get('https://httpbin.org/ip', proxies=proxies)
print(response.json())

Настройка SOCKS5-прокси:

Для работы с SOCKS5 требуется дополнительная библиотека requests[socks] или PySocks. Установка:

pip install requests[socks]

Пример использования SOCKS5:

import requests

# SOCKS5 без аутентификации
proxies = {
    'http': 'socks5://123.45.67.89:1080',
    'https': 'socks5://123.45.67.89:1080'
}

# SOCKS5 с аутентификацией
proxies_auth = {
    'http': 'socks5://user:pass@123.45.67.89:1080',
    'https': 'socks5://user:pass@123.45.67.89:1080'
}

response = requests.get('https://httpbin.org/ip', proxies=proxies_auth)
print(response.json())

SOCKS5 прокси особенно полезны при работе с резидентными прокси, так как этот протокол обеспечивает более надёжное туннелирование трафика и поддерживает UDP (необходимо для некоторых приложений).

Ротация прокси в requests

При парсинге больших объёмов данных использование одного IP-адреса приводит к блокировкам. Ротация прокси — это циклическая смена IP для распределения нагрузки и обхода rate limits.

Простая ротация через список:

import requests
import itertools

# Список прокси-серверов
proxy_list = [
    'http://user:pass@proxy1.example.com:8080',
    'http://user:pass@proxy2.example.com:8080',
    'http://user:pass@proxy3.example.com:8080',
]

# Создание бесконечного итератора
proxy_pool = itertools.cycle(proxy_list)

# Выполнение запросов с ротацией
for i in range(10):
    proxy = next(proxy_pool)
    proxies = {'http': proxy, 'https': proxy}
    
    try:
        response = requests.get('https://httpbin.org/ip', proxies=proxies, timeout=5)
        print(f"Запрос {i+1}: IP = {response.json()['origin']}")
    except Exception as e:
        print(f"Ошибка с прокси {proxy}: {e}")

Ротация с сессиями для сохранения cookies:

import requests
from itertools import cycle

class ProxyRotator:
    def __init__(self, proxy_list):
        self.proxy_pool = cycle(proxy_list)
        self.session = requests.Session()
    
    def get(self, url, **kwargs):
        proxy = next(self.proxy_pool)
        self.session.proxies = {'http': proxy, 'https': proxy}
        return self.session.get(url, **kwargs)

# Использование
proxy_list = [
    'http://user:pass@proxy1.example.com:8080',
    'http://user:pass@proxy2.example.com:8080',
]

rotator = ProxyRotator(proxy_list)

for i in range(5):
    response = rotator.get('https://httpbin.org/ip', timeout=5)
    print(f"Запрос {i+1}: {response.json()['origin']}")

Случайная ротация для непредсказуемости:

import requests
import random

proxy_list = [
    'http://user:pass@proxy1.example.com:8080',
    'http://user:pass@proxy2.example.com:8080',
    'http://user:pass@proxy3.example.com:8080',
]

def get_random_proxy():
    proxy = random.choice(proxy_list)
    return {'http': proxy, 'https': proxy}

# Каждый запрос с случайным прокси
for i in range(5):
    response = requests.get('https://httpbin.org/ip', proxies=get_random_proxy(), timeout=5)
    print(f"Запрос {i+1}: {response.json()['origin']}")

Случайная ротация эффективнее при работе с сайтами, которые отслеживают паттерны запросов. Последовательная смена IP может выглядеть подозрительно, тогда как случайный выбор имитирует поведение разных пользователей.

Настройка прокси в aiohttp

Библиотека aiohttp предназначена для асинхронных HTTP-запросов и критична для высоконагруженных парсеров. Настройка прокси отличается от requests — используется параметр proxy (в единственном числе).

Базовый пример с HTTP-прокси:

import aiohttp
import asyncio

async def fetch_with_proxy():
    proxy = 'http://123.45.67.89:8080'
    
    async with aiohttp.ClientSession() as session:
        async with session.get('https://httpbin.org/ip', proxy=proxy) as response:
            data = await response.json()
            print(data)

# Запуск
asyncio.run(fetch_with_proxy())

Прокси с аутентификацией:

В aiohttp аутентификация передаётся через объект aiohttp.BasicAuth или прямо в URL:

import aiohttp
import asyncio

async def fetch_with_auth_proxy():
    # Вариант 1: Учётные данные в URL
    proxy = 'http://user123:pass456@proxy.example.com:8080'
    
    async with aiohttp.ClientSession() as session:
        async with session.get('https://httpbin.org/ip', proxy=proxy) as response:
            print(await response.json())

# Вариант 2: Через BasicAuth (для некоторых прокси)
async def fetch_with_basic_auth():
    proxy = 'http://proxy.example.com:8080'
    proxy_auth = aiohttp.BasicAuth('user123', 'pass456')
    
    async with aiohttp.ClientSession() as session:
        async with session.get('https://httpbin.org/ip', 
                                proxy=proxy, 
                                proxy_auth=proxy_auth) as response:
            print(await response.json())

asyncio.run(fetch_with_auth_proxy())

SOCKS5 в aiohttp:

Для SOCKS5 требуется библиотека aiohttp-socks:

pip install aiohttp-socks

import asyncio
from aiohttp_socks import ProxyConnector
import aiohttp

async def fetch_with_socks5():
    connector = ProxyConnector.from_url('socks5://user:pass@123.45.67.89:1080')
    
    async with aiohttp.ClientSession(connector=connector) as session:
        async with session.get('https://httpbin.org/ip') as response:
            print(await response.json())

asyncio.run(fetch_with_socks5())

При работе с мобильными прокси для парсинга социальных сетей или маркетплейсов рекомендуется использовать именно aiohttp — асинхронность позволяет обрабатывать сотни запросов параллельно без блокировки потока выполнения.

Асинхронная ротация и пул прокси

Для высоконагруженных парсеров критична эффективная ротация прокси с обработкой отказов и автоматической заменой нерабочих IP. Рассмотрим продвинутые паттерны для aiohttp.

Класс для управления пулом прокси:

import aiohttp
import asyncio
from itertools import cycle
from typing import List, Optional

class ProxyPool:
    def __init__(self, proxy_list: List[str]):
        self.proxy_list = proxy_list
        self.proxy_cycle = cycle(proxy_list)
        self.failed_proxies = set()
    
    def get_next_proxy(self) -> Optional[str]:
        """Получить следующий рабочий прокси"""
        for _ in range(len(self.proxy_list)):
            proxy = next(self.proxy_cycle)
            if proxy not in self.failed_proxies:
                return proxy
        return None  # Все прокси недоступны
    
    def mark_failed(self, proxy: str):
        """Отметить прокси как нерабочий"""
        self.failed_proxies.add(proxy)
        print(f"Прокси {proxy} помечен как недоступный")
    
    async def fetch(self, session: aiohttp.ClientSession, url: str, **kwargs):
        """Выполнить запрос с автоматической сменой прокси при ошибке"""
        max_retries = 3
        
        for attempt in range(max_retries):
            proxy = self.get_next_proxy()
            if not proxy:
                raise Exception("Все прокси недоступны")
            
            try:
                async with session.get(url, proxy=proxy, timeout=aiohttp.ClientTimeout(total=10), **kwargs) as response:
                    return await response.json()
            except (aiohttp.ClientError, asyncio.TimeoutError) as e:
                print(f"Ошибка с прокси {proxy}: {e}")
                self.mark_failed(proxy)
                continue
        
        raise Exception(f"Не удалось выполнить запрос после {max_retries} попыток")

# Использование
async def main():
    proxy_list = [
        'http://user:pass@proxy1.example.com:8080',
        'http://user:pass@proxy2.example.com:8080',
        'http://user:pass@proxy3.example.com:8080',
    ]
    
    pool = ProxyPool(proxy_list)
    
    async with aiohttp.ClientSession() as session:
        # Выполнение 10 запросов с автоматической ротацией
        tasks = [pool.fetch(session, 'https://httpbin.org/ip') for _ in range(10)]
        results = await asyncio.gather(*tasks, return_exceptions=True)
        
        for i, result in enumerate(results):
            if isinstance(result, Exception):
                print(f"Запрос {i+1} завершился ошибкой: {result}")
            else:
                print(f"Запрос {i+1}: IP = {result.get('origin')}")

asyncio.run(main())

Параллельная обработка с ограничением concurrency:

import aiohttp
import asyncio
from itertools import cycle

async def fetch_url(session, url, proxy, semaphore):
    async with semaphore:  # Ограничение одновременных запросов
        try:
            async with session.get(url, proxy=proxy, timeout=aiohttp.ClientTimeout(total=10)) as response:
                data = await response.json()
                return {'url': url, 'ip': data.get('origin'), 'status': response.status}
        except Exception as e:
            return {'url': url, 'error': str(e)}

async def main():
    urls = [f'https://httpbin.org/ip' for _ in range(50)]  # 50 запросов
    proxy_list = [
        'http://user:pass@proxy1.example.com:8080',
        'http://user:pass@proxy2.example.com:8080',
    ]
    proxy_cycle = cycle(proxy_list)
    
    # Ограничение: не более 10 одновременных запросов
    semaphore = asyncio.Semaphore(10)
    
    async with aiohttp.ClientSession() as session:
        tasks = [
            fetch_url(session, url, next(proxy_cycle), semaphore)
            for url in urls
        ]
        results = await asyncio.gather(*tasks)
        
        # Анализ результатов
        successful = [r for r in results if 'ip' in r]
        failed = [r for r in results if 'error' in r]
        
        print(f"Успешных запросов: {len(successful)}")
        print(f"Неудачных запросов: {len(failed)}")

asyncio.run(main())

Использование asyncio.Semaphore критически важно при работе с прокси — слишком большое количество одновременных соединений через один IP может вызвать блокировку со стороны целевого сайта или прокси-провайдера.

Обработка ошибок и таймаутов

Работа с прокси сопряжена с повышенным количеством ошибок: таймауты, разрывы соединения, отказы прокси-серверов. Правильная обработка ошибок — ключ к стабильности парсера.

Типичные ошибки при работе с прокси:

Ошибка	Причина	Решение
`ProxyError`	Прокси-сервер недоступен	Переключиться на другой прокси
`ConnectTimeout`	Прокси не отвечает вовремя	Увеличить таймаут или сменить прокси
`ProxyAuthenticationRequired`	Неверные логин/пароль	Проверить учётные данные
`SSLError`	Проблемы с SSL-сертификатом	Отключить проверку SSL (не рекомендуется)
`TooManyRedirects`	Прокси создаёт редирект-петлю	Сменить прокси или ограничить редиректы

Обработка ошибок в requests:

import requests
from requests.exceptions import ProxyError, ConnectTimeout, RequestException

def fetch_with_retry(url, proxies, max_retries=3):
    for attempt in range(max_retries):
        try:
            response = requests.get(
                url, 
                proxies=proxies, 
                timeout=(5, 10),  # (connect timeout, read timeout)
                allow_redirects=True,
                verify=True  # Проверка SSL-сертификата
            )
            response.raise_for_status()  # Вызовет исключение при 4xx/5xx
            return response.json()
            
        except ProxyError as e:
            print(f"Попытка {attempt + 1}: Прокси недоступен - {e}")
        except ConnectTimeout as e:
            print(f"Попытка {attempt + 1}: Таймаут подключения - {e}")
        except requests.exceptions.HTTPError as e:
            print(f"Попытка {attempt + 1}: HTTP ошибка {e.response.status_code}")
            if e.response.status_code == 407:  # Proxy Authentication Required
                print("Ошибка аутентификации прокси!")
                break  # Не повторять при ошибке авторизации
        except RequestException as e:
            print(f"Попытка {attempt + 1}: Общая ошибка - {e}")
        
        if attempt < max_retries - 1:
            print(f"Повтор через 2 секунды...")
            import time
            time.sleep(2)
    
    raise Exception(f"Не удалось выполнить запрос после {max_retries} попыток")

# Использование
proxies = {'http': 'http://user:pass@proxy.example.com:8080', 'https': 'http://user:pass@proxy.example.com:8080'}
try:
    data = fetch_with_retry('https://httpbin.org/ip', proxies)
    print(data)
except Exception as e:
    print(f"Критическая ошибка: {e}")

Обработка ошибок в aiohttp:

import aiohttp
import asyncio
from aiohttp import ClientError, ClientProxyConnectionError

async def fetch_with_retry(session, url, proxy, max_retries=3):
    for attempt in range(max_retries):
        try:
            timeout = aiohttp.ClientTimeout(total=10, connect=5)
            async with session.get(url, proxy=proxy, timeout=timeout) as response:
                response.raise_for_status()
                return await response.json()
                
        except ClientProxyConnectionError as e:
            print(f"Попытка {attempt + 1}: Ошибка подключения к прокси - {e}")
        except asyncio.TimeoutError:
            print(f"Попытка {attempt + 1}: Таймаут")
        except aiohttp.ClientHttpProxyError as e:
            print(f"Попытка {attempt + 1}: HTTP ошибка прокси - {e}")
            if e.status == 407:
                print("Ошибка аутентификации прокси!")
                break
        except ClientError as e:
            print(f"Попытка {attempt + 1}: Общая ошибка клиента - {e}")
        
        if attempt < max_retries - 1:
            await asyncio.sleep(2)
    
    raise Exception(f"Не удалось выполнить запрос после {max_retries} попыток")

async def main():
    proxy = 'http://user:pass@proxy.example.com:8080'
    async with aiohttp.ClientSession() as session:
        try:
            data = await fetch_with_retry(session, 'https://httpbin.org/ip', proxy)
            print(data)
        except Exception as e:
            print(f"Критическая ошибка: {e}")

asyncio.run(main())

Настройка таймаутов:

Правильная настройка таймаутов критична для стабильности. Рекомендуемые значения:

Connect timeout: 5-10 секунд (время на установку соединения с прокси)
Read timeout: 10-30 секунд (время на получение ответа от целевого сайта)
Total timeout: 30-60 секунд (общее время запроса)

Для медленных резидентных прокси рекомендуется увеличивать таймауты до 20-30 секунд на соединение, так как маршрутизация через реальных провайдеров может занимать больше времени.

Лучшие практики и оптимизация

Эффективная работа с прокси требует соблюдения набора правил для минимизации блокировок и максимизации производительности.

1. Использование Session для переиспользования соединений:

# requests: Session переиспользует TCP-соединения
session = requests.Session()
session.proxies = {'http': proxy, 'https': proxy}

for url in urls:
    response = session.get(url)  # Быстрее, чем requests.get()

# aiohttp: Session обязателен для асинхронности
async with aiohttp.ClientSession() as session:
    tasks = [session.get(url, proxy=proxy) for url in urls]
    await asyncio.gather(*tasks)

2. Установка реалистичных User-Agent и заголовков:

import requests

headers = {
    'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36',
    'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8',
    'Accept-Language': 'en-US,en;q=0.5',
    'Accept-Encoding': 'gzip, deflate, br',
    'DNT': '1',
    'Connection': 'keep-alive',
    'Upgrade-Insecure-Requests': '1'
}

proxies = {'http': proxy, 'https': proxy}
response = requests.get('https://example.com', headers=headers, proxies=proxies)

3. Ограничение rate limit (запросов в секунду):

import time
import requests

class RateLimiter:
    def __init__(self, max_requests_per_second):
        self.max_requests = max_requests_per_second
        self.interval = 1.0 / max_requests_per_second
        self.last_request_time = 0
    
    def wait(self):
        elapsed = time.time() - self.last_request_time
        if elapsed < self.interval:
            time.sleep(self.interval - elapsed)
        self.last_request_time = time.time()

# Использование: не более 2 запросов в секунду
limiter = RateLimiter(2)
proxies = {'http': proxy, 'https': proxy}

for url in urls:
    limiter.wait()
    response = requests.get(url, proxies=proxies)

4. Логирование и мониторинг прокси:

import logging
from collections import defaultdict

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class ProxyMonitor:
    def __init__(self):
        self.stats = defaultdict(lambda: {'success': 0, 'failed': 0, 'total_time': 0})
    
    def log_request(self, proxy, success, response_time):
        stats = self.stats[proxy]
        if success:
            stats['success'] += 1
        else:
            stats['failed'] += 1
        stats['total_time'] += response_time
        
        # Логирование каждые 10 запросов
        total = stats['success'] + stats['failed']
        if total % 10 == 0:
            avg_time = stats['total_time'] / total
            success_rate = stats['success'] / total * 100
            logger.info(f"Прокси {proxy}: {total} запросов, успех {success_rate:.1f}%, avg {avg_time:.2f}s")

monitor = ProxyMonitor()

# В коде запроса
import time
start = time.time()
try:
    response = requests.get(url, proxies=proxies, timeout=10)
    monitor.log_request(proxy, True, time.time() - start)
except Exception as e:
    monitor.log_request(proxy, False, time.time() - start)
    logger.error(f"Ошибка с прокси {proxy}: {e}")

5. Кэширование DNS для ускорения:

# aiohttp с кэшированием DNS
import aiohttp
from aiohttp.resolver import AsyncResolver

resolver = AsyncResolver(nameservers=['8.8.8.8', '8.8.4.4'])
connector = aiohttp.TCPConnector(resolver=resolver, ttl_dns_cache=300)

async with aiohttp.ClientSession(connector=connector) as session:
    # Запросы будут использовать кэш DNS на 5 минут
    async with session.get(url, proxy=proxy) as response:
        data = await response.json()

6. Обработка капчи и блокировок:

Совет: При получении статуса 403, 429 или капчи рекомендуется:

Сменить прокси на IP из другой подсети
Увеличить задержку между запросами (до 5-10 секунд)
Изменить User-Agent и другие заголовки
Использовать cookies из предыдущих успешных сессий

Сравнение requests и aiohttp для прокси

Выбор между requests и aiohttp зависит от задачи и объёма данных. Рассмотрим ключевые различия.

Критерий	requests	aiohttp
Синхронность	Синхронный (блокирующий)	Асинхронный (неблокирующий)
Производительность	~10-50 запросов/сек	~100-1000 запросов/сек
Простота кода	Проще для новичков	Требует знания async/await
Настройка прокси	Словарь `proxies`	Параметр `proxy`
SOCKS5 поддержка	Через `requests[socks]`	Через `aiohttp-socks`
Использование памяти	Меньше (один поток)	Больше (множество задач)
Лучше для	Простые скрипты, <100 запросов	Парсеры, >1000 запросов

Когда использовать requests:

Простые скрипты для разовых задач
Прототипирование и тестирование
Небольшой объём запросов (до 100 в минуту)
Когда важна простота кода и читаемость
Интеграция с синхронными библиотеками

Когда использовать aiohttp:

Парсинг больших объёмов данных (тысячи страниц)
Мониторинг множества источников в реальном времени
API-сервисы с высокой нагрузкой
Когда критична скорость обработки
Работа с WebSocket через прокси

Практическое сравнение производительности:

# Тест: 100 запросов через прокси

# requests (синхронный) - ~50 секунд
import requests
import time

start = time.time()
proxies = {'http': proxy, 'https': proxy}
for i in range(100):
    response = requests.get('https://httpbin.org/ip', proxies=proxies)
print(f"requests: {time.time() - start:.2f} секунд")

# aiohttp (асинхронный) - ~5 секунд
import aiohttp
import asyncio

async def fetch_all():
    async with aiohttp.ClientSession() as session:
        tasks = [
            session.get('https://httpbin.org/ip', proxy=proxy)
            for _ in range(100)
        ]
        await asyncio.gather(*tasks)

start = time.time()
asyncio.run(fetch_all())
print(f"aiohttp: {time.time() - start:.2f} секунд")

При использовании прокси дата-центров для парсинга с высокой скоростью aiohttp показывает преимущество в 10-20 раз по сравнению с requests благодаря параллельной обработке запросов.

Заключение

Настройка прокси в Python через библиотеки requests и aiohttp — это фундаментальный навык для разработки парсеров, автоматизации сбора данных и обхода географических ограничений. Библиотека requests подходит для простых скриптов и прототипирования благодаря понятному синхронному API, тогда как aiohttp обеспечивает высокую производительность при обработке тысяч запросов через асинхронную архитектуру.

Ключевые моменты для эффективной работы с прокси в Python: правильная обработка ошибок и таймаутов, реализация ротации IP-адресов для распределения нагрузки, использование Session для переиспользования соединений, настройка реалистичных заголовков и User-Agent, мониторинг производительности прокси-серверов. Для SOCKS5 прокси требуются дополнительные библиотеки — requests[socks] или aiohttp-socks.

При выборе типа прокси для парсинга учитывайте специфику задачи: для высоконагруженных парсеров с тысячами запросов подойдут быстрые прокси дата-центров, для обхода жёстких антибот-систем и работы с социальными сетями рекомендуются резидентные прокси с реальными IP домашних пользователей, а для задач, требующих максимальной анонимности и имитации мобильного трафика, оптимальны мобильные прокси с IP операторов сотовой связи.

Если вы планируете разрабатывать высокопроизводительные парсеры или автоматизировать сбор данных из множества источников, рекомендуем попробовать резидентные прокси — они обеспечивают высокий уровень анонимности, минимальный риск блокировок и стабильную работу с большинством защищённых веб-сервисов. Для технических задач с высокой скоростью обработки также подойдут прокси дата-центров с низкой задержкой и высокой пропускной способностью.

Настройка прокси в Python requests и aiohttp: полное руководство с примерами кода