Если вы парсите, краулите или автоматизируете что-либо в Python в сколько-нибудь серьёзных масштабах, рано или поздно целевой сайт замечает, что все запросы идут с одного IP, и закрывает вам доступ. Ротация прокси в Python распределяет трафик по множеству IP-адресов, чтобы ни один из них не упёрся в rate limit и не схлопотал бан. В этом руководстве — готовый код для requests, httpx и aiohttp, а также как выбрать между ротационным шлюзом и управляемым пулом IP.
Два сквозных решения определяют всё, что ниже. Первое: ротация против sticky-сессий — новый IP на каждый запрос или один IP, удерживаемый на весь многошаговый сценарий. Второе: резидентные прокси против дата-центровых — см. матрицу выбора типа прокси. Разберитесь с этими двумя вопросами, и код станет самой простой частью.
1. Ротация на шлюзе (рекомендуется). Вы подключаетесь к одному эндпоинту, а провайдер выдаёт вам свежий исходящий IP на каждое новое соединение. Никаких списков для управления, никаких мёртвых IP для отсева. Пул динамических резидентных прокси JIBAO работает именно так: тот же хост и порт, новый IP на каждый запрос.
2. Самостоятельно управляемый список IP. Вы держите список записей host:port и выбираете один на каждый запрос. Больше контроля, но проверка работоспособности, ретраи и отслеживание банов целиком на вас. Полезно с выделенными дата-центровыми IP, когда нужны известные, стабильные адреса.
Простейший случай — пропустить запрос через шлюз, который ротирует IP за вас. Используйте socks5h:// (завершающая h означает, что DNS разрешается на стороне прокси, что исключает утечку запросов):
import requests
# Ротационный резидентный шлюз JIBAO: новый исходящий IP на каждое соединение
PROXY = "socks5h://USERNAME:[email protected]:10001"
proxies = {"http": PROXY, "https": PROXY}
for _ in range(5):
r = requests.get("https://httpbin.org/ip", proxies=proxies, timeout=30)
print(r.json()["origin"]) # разный IP на каждой итерации
Нужно, чтобы requests умел SOCKS5? Поставьте extra: pip install "requests[socks]". Если предпочитаете HTTP-прокси, поменяйте схему на http:// — всё остальное идентично.
import random
import requests
PROXY_POOL = [
"socks5h://USERNAME-country-us:[email protected]:10001",
"socks5h://USERNAME-country-uk:[email protected]:10001",
"socks5h://USERNAME-country-de:[email protected]:10001",
]
def fetch(url):
proxy = random.choice(PROXY_POOL)
return requests.get(url, proxies={"http": proxy, "https": proxy}, timeout=30)
resp = fetch("https://example.com")
print(resp.status_code)
Сценарии входа, корзины и постраничная выдача ломаются, если IP меняется посреди задачи. Закрепите IP, добавив токен сессии в имя пользователя; переиспользуйте тот же токен, чтобы сохранить тот же исходящий IP на всё время его жизни:
import requests
SESSION_ID = "task-0001"
PROXY = f"socks5h://USERNAME-session-{SESSION_ID}:[email protected]:10001"
s = requests.Session()
s.proxies = {"http": PROXY, "https": PROXY}
s.post("https://example.com/login", data={"u": "me", "p": "secret"})
s.get("https://example.com/dashboard") # тот же IP, сессия не нарушена
httpx даёт современный клиент с HTTP/2 и нативным async. Учтите, что в новых версиях передаётся единственный аргумент proxy=:
import httpx
PROXY = "socks5h://USERNAME:[email protected]:10001"
with httpx.Client(proxy=PROXY, timeout=30) as client:
for _ in range(5):
print(client.get("https://httpbin.org/ip").json()["origin"])
Поддержка SOCKS в httpx требует pip install "httpx[socks]". Для HTTP/2 добавьте http2=True в клиент — так ваш трафик становится больше похож на настоящий браузер, что хорошо сочетается с подгонкой TLS-отпечатка.
Для высокой пропускной способности запускайте множество запросов одновременно. Каждое соединение через шлюз получает свой исходящий IP, так что параллелизм и ротация идут рука об руку:
import asyncio
import aiohttp
from aiohttp_socks import ProxyConnector
PROXY = "socks5://USERNAME:[email protected]:10001"
async def fetch(url):
connector = ProxyConnector.from_url(PROXY)
async with aiohttp.ClientSession(connector=connector) as session:
async with session.get(url, timeout=aiohttp.ClientTimeout(total=30)) as r:
return await r.text()
async def main():
urls = ["https://httpbin.org/ip"] * 20
results = await asyncio.gather(*(fetch(u) for u in urls))
print(f"fetched {len(results)} pages")
asyncio.run(main())
Установите SOCKS-коннектор через pip install aiohttp_socks. Создавайте свежий коннектор на каждую задачу, чтобы каждый запрос открывал новое проксированное соединение.
Ротация окупается только тогда, когда вы повторяете неудачные запросы с нового IP. Отступайте экспоненциально и меняйте токен сессии при каждой попытке:
import requests
from time import sleep
def fetch_with_retry(url, max_retries=4):
for attempt in range(max_retries):
proxy = f"socks5h://USERNAME-session-r{attempt}:[email protected]:10001"
try:
r = requests.get(url, proxies={"http": proxy, "https": proxy}, timeout=30)
if r.status_code in (403, 429):
raise requests.HTTPError(f"blocked: {r.status_code}")
r.raise_for_status()
return r.text
except requests.RequestException:
sleep(2 ** attempt) # 1s, 2s, 4s, 8s
raise RuntimeError(f"failed after {max_retries} tries: {url}")
Перед большим прогоном убедитесь, что IP реально меняется. Стукнитесь к echo-эндпоинту несколько раз и проверьте, что IP различаются. Чтобы заодно убедиться, что вас не выдаёт TLS-отпечаток, прогоните запрос через бесплатный чекер JA3/TLS-отпечатка от JIBAO — ротируемый IP с очевидно «питоновским» отпечатком всё равно получит блок.
Когда ротация отлажена, следующая стена, в которую вы упрётесь, обычно не IP, а фингерпринтинг. Если чистые резидентные IP всё равно ловят 403, прочитайте, как обойти TLS-фингерпринтинг с помощью curl_cffi, и полный рецепт обхода Cloudflare.
Получите $5 бесплатного баланса и ротационный резидентный шлюз, выдающий новый IP на каждый запрос.
Начать бесплатноНовым пользователям — 5U при регистрации, бонус к первому пополнению. Акция ограничена по времени.