Прокси в Python: requests, httpx, aiohttp и Scrapy

Формат адреса прокси

Все библиотеки ниже понимают прокси в виде URL: схема, логин и пароль, хост и порт.

http://login:[email protected]:8000
socks5://login:[email protected]:1080

В кабинете TrueSocks данные доступа можно скопировать в разных форматах, а список вида ip:port:login:password переводится в URL в конвертере прокси. Адреса в примерах условные: подставьте свои.

Обратите внимание на схему у HTTP(S)-прокси: даже для запросов к сайтам по HTTPS адрес самого прокси обычно начинается с http://. Клиент подключается к прокси по HTTP и просит его открыть туннель методом CONNECT, а шифрование между вами и сайтом при этом сохраняется. Подробнее о протоколах в статье HTTP(S) и SOCKS5.

requests

import requests

proxy = "http://login:[email protected]:8000"
proxies = {"http": proxy, "https": proxy}

r = requests.get("https://httpbin.org/ip", proxies=proxies, timeout=15)
print(r.json())

Ключи словаря http и https означают схему сайта, а не прокси: так вы говорите, через какой прокси ходить к HTTP- и HTTPS-сайтам. Обязательно задавайте timeout: без него запрос через недоступный прокси может висеть очень долго.

SOCKS5 в requests

proxy = "socks5h://login:[email protected]:1080"
proxies = {"http": proxy, "https": proxy}
r = requests.get("https://httpbin.org/ip", proxies=proxies, timeout=15)

Буква h в схеме socks5h значит, что имя сайта разрешается в IP на стороне прокси. Со схемой socks5 DNS-запрос делает ваш компьютер, и это может раскрыть, какие сайты вы открываете, вашему провайдеру.

Сессия и переменные окружения

Для серии запросов используйте requests.Session(): соединения переиспользуются, и это заметно быстрее. Учтите, что requests читает переменные окружения HTTP_PROXY и HTTPS_PROXY. Чтобы они не вмешивались, передавайте proxies явно в каждый вызов или отключите их чтение через session.trust_env = False.

httpx

import httpx

proxy = "http://login:[email protected]:8000"

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

async def main():
    async with httpx.AsyncClient(proxy=proxy, timeout=15) as client:
        r = await client.get("https://httpbin.org/ip")
        print(r.json())

Параметр proxy есть в актуальных версиях httpx. В старых версиях использовался параметр proxies, в новых он удалён, поэтому при обновлении библиотеки проверьте код. Для SOCKS5 установите pip install "httpx[socks]" и укажите адрес со схемой socks5://.

Прокси через переменные окружения

Если код менять неудобно, например нужно пустить через прокси чужой скрипт или утилиту, прокси можно задать переменными окружения. Их понимают requests, httpx, pip, curl и многие другие программы.

export HTTP_PROXY="http://login:[email protected]:8000"
export HTTPS_PROXY="http://login:[email protected]:8000"
export NO_PROXY="localhost,127.0.0.1"
python script.py

В PowerShell на Windows то же самое задаётся командой $env:HTTPS_PROXY = "http://login:[email protected]:8000". Переменная NO_PROXY перечисляет адреса, которые должны идти напрямую, например локальные сервисы и базу данных. Учтите обратную сторону: если переменные заданы глобально, через прокси пойдут и запросы, которым он не нужен, включая установку пакетов.

aiohttp по умолчанию переменные окружения не читает. Чтобы он их учитывал, создайте сессию с параметром trust_env=True.

aiohttp

import aiohttp

async def fetch(url):
    timeout = aiohttp.ClientTimeout(total=20)
    async with aiohttp.ClientSession(timeout=timeout) as session:
        async with session.get(
            url,
            proxy="http://203.0.113.10:8000",
            proxy_auth=aiohttp.BasicAuth("login", "password"),
        ) as resp:
            return await resp.text()

Логин и пароль можно передать отдельно через proxy_auth или прямо в адресе прокси. Встроенной поддержки SOCKS в aiohttp нет, для неё используют пакет aiohttp-socks: он подменяет коннектор сессии.

from aiohttp_socks import ProxyConnector

connector = ProxyConnector.from_url("socks5://login:[email protected]:1080")
async with aiohttp.ClientSession(connector=connector) as session:
    async with session.get("https://httpbin.org/ip") as resp:
        print(await resp.text())

В асинхронном коде легко отправить сотни запросов одновременно. Ограничивайте параллельность, например через asyncio.Semaphore, иначе вы упрётесь и в лимиты сайта, и в возможности канала.

Scrapy

В Scrapy прокси задаётся для каждого запроса через meta. Встроенный HttpProxyMiddleware включён по умолчанию: он сам берёт логин и пароль из адреса и добавляет заголовок авторизации.

yield scrapy.Request(
    url,
    callback=self.parse,
    meta={"proxy": "http://login:[email protected]:8000"},
)

Чтобы распределять запросы по списку адресов, напишите небольшой middleware. Его номер должен быть меньше 750, чтобы он сработал раньше встроенного HttpProxyMiddleware.

import random

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

    @classmethod
    def from_crawler(cls, crawler):
        return cls(crawler.settings.getlist("PROXY_LIST"))

    def process_request(self, request, spider):
        request.meta.setdefault("proxy", random.choice(self.proxies))

# settings.py
PROXY_LIST = ["http://login:[email protected]:8000", "http://login:[email protected]:8000"]
DOWNLOADER_MIDDLEWARES = {"myproject.middlewares.RotateProxyMiddleware": 350}
AUTOTHROTTLE_ENABLED = True
CONCURRENT_REQUESTS_PER_DOMAIN = 4

Scrapy из коробки работает с HTTP(S)-прокси, а SOCKS5 не поддерживает, поэтому для него заказывайте прокси с протоколом HTTP(S). Встроенный AutoThrottle сам подстраивает темп под скорость ответов сайта: включайте его в любом проекте, который работает через прокси.

Ещё два полезных параметра: RETRY_HTTP_CODES определяет, при каких ответах сервера Scrapy повторяет запрос, а DOWNLOAD_TIMEOUT ограничивает время ожидания ответа. Не добавляйте 403 в список повторов без разбора: если сайт отказывает в доступе, повторные запросы с того же адреса только ухудшат ситуацию. Для браузерных сценариев с JavaScript, например через Playwright, прокси задаётся при запуске браузера или создании контекста, и там тоже лучше один контекст на один адрес.

Ротация по списку и обработка ошибок

Серверные IPv4 и ISP-адреса статичны: если нужно распределять запросы по адресам, это делает ваш код. Самый простой способ - перебирать список по кругу.

import itertools, time, requests

pool = itertools.cycle(open("proxies.txt").read().split())

def get(url):
    for _ in range(3):
        proxy = next(pool)
        try:
            r = requests.get(url, proxies={"http": proxy, "https": proxy}, timeout=15)
        except requests.exceptions.ProxyError:
            continue
        if r.status_code == 429:
            time.sleep(int(r.headers.get("Retry-After", 30)))
            continue
        return r
    return None

Проверьте список до запуска

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

from concurrent.futures import ThreadPoolExecutor
import requests

def check(proxy):
    try:
        r = requests.get("https://httpbin.org/ip",
                         proxies={"http": proxy, "https": proxy}, timeout=10)
        return proxy, r.json()["origin"], round(r.elapsed.total_seconds(), 2)
    except Exception as e:
        return proxy, None, type(e).__name__

proxies = open("proxies.txt").read().split()
with ThreadPoolExecutor(max_workers=10) as ex:
    for proxy, ip, info in ex.map(check, proxies):
        print(ip or "FAIL", info)

Скрипт показывает внешний IP и время ответа для каждого адреса. Если адрес не отвечает, сначала сверьте протокол и порт: прокси, заказанный как HTTP(S), не будет работать со схемой socks5://, и наоборот. Если адрес из заказа не работает, его можно бесплатно заменить в течение 24 часов после покупки. Подробнее о проверке страны, скорости и утечек в статье как проверить прокси.

Где хранить данные доступа

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

• Не пишите пароли в код. Храните список прокси в отдельном файле или в переменных окружения и добавьте файл в .gitignore.
• Разделяйте окружения. Для разработки и для боевого запуска удобно держать разные списки адресов, чтобы тесты не расходовали рабочий пул.
• Не логируйте адрес целиком. В логах достаточно хоста и порта, пароль туда попадать не должен.
• Используйте привязку по IP на сервере. Если скрипт работает на сервере с постоянным адресом, авторизация по IP избавляет от хранения пароля. Сравнение способов в статье логин или привязка по IP.

Если вы работаете с большим количеством адресов, удобнее получать актуальный список через API, а не переписывать файл вручную после каждого продления или замены.

Сколько потоков на один адрес

Универсального числа нет: оно зависит от сайта, а не от прокси. Начинайте с одного-двух одновременных запросов на адрес и небольших пауз, следите за долей ответов 429 и временем ответа и увеличивайте нагрузку постепенно. Если ошибок становится больше, добавляйте адреса в пул, а не потоки на каждый адрес. Как понять, почему скорость падает, рассказано в статье почему прокси работает медленно.

Какие прокси брать для Python-проектов

• Сбор открытых данных: прокси для парсинга или IPv4 Эконом - много адресов нужной страны по низкой цене.
• Сайты с поддержкой IPv6: IPv6 Эконом обходятся в разы дешевле. Как проверить поддержку, рассказано в статье IPv4 или IPv6.
• Сайты, которые ограничивают серверные подсети: резидентские прокси с ротацией или липкой сессией.
• Большие пулы: IPv4 Опт со скидками за объём и выгрузкой списка через API.

Больше о сценарии на странице прокси для парсинга.