Практическое руководство по SOCKS5-прокси в Python requestsАутентификация, DNS, повторы и ротация
Полное руководство: использование SOCKS5-прокси в библиотеке Python requests, включая аутентификацию, разрешение DNS, обработку ошибок и практические паттерны ротации прокси.
Быстрый старт: минимальный пример SOCKS5
Сначала убедитесь, что вам действительно выдали именно SOCKS5-прокси, а не HTTP-прокси, иначе приведённый ниже код не сможет установить соединение.
import requests
# For credentials with special chars, use URL encoding:
# from urllib.parse import quote
# PROXY = f"socks5h://{quote('user@123')}:{quote('p@ss!')}@host:port"
PROXY = "socks5h://username:password@host:port" # e.g. socks5h://user:pass@1.2.3.4:1080
proxies = {"http": PROXY, "https": PROXY}
resp = requests.get("https://httpbin.org/ip", proxies=proxies, timeout=20)
resp.raise_for_status()
print(resp.json())Важные замечания
- • Установите поддержку SOCKS командой
pip install "requests[socks]"(PySocks установится автоматически) - •
socks5hгарантирует, что DNS разрешает прокси. Обычныйsocks5разрешает DNS локально
Рекомендуемый способ создания Session для продакшена: тайм-ауты + повторы (с экспоненциальной задержкой)
Используйте Session с примонтированным адаптером на базе urllib3.Retry для эффективного повторного использования соединений:
import os
from urllib.parse import quote
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
# Build a socks5h proxy URL (remote DNS)
user = os.getenv("PROXY_USER")
pw = os.getenv("PROXY_PASS", "")
host = os.getenv("PROXY_HOST", "127.0.0.1")
port = int(os.getenv("PROXY_PORT", "1080"))
PROXY = (
f"socks5h://{quote(user)}:{quote(pw)}@{host}:{port}"
if user else f"socks5h://{host}:{port}"
)
session = requests.Session()
session.trust_env = False # optional: ignore env proxies (HTTP_PROXY/HTTPS_PROXY)
session.proxies.update({"http": PROXY, "https": PROXY})
retries = Retry(
total=5,
backoff_factor=0.5, # 0.5, 1, 2, 4... seconds
status_forcelist=[429, 500, 502, 503, 504],
allowed_methods=frozenset(["HEAD", "GET", "OPTIONS"]),
respect_retry_after_header=True, # explicit for portability with urllib3 v1.26+
)
adapter = HTTPAdapter(max_retries=retries, pool_connections=50, pool_maxsize=50)
session.mount("http://", adapter)
session.mount("https://", adapter)
def fetch(url):
r = session.get(url, timeout=(5, 30)) # (connect, read)
r.raise_for_status()
return r
if __name__ == "__main__":
print(fetch("https://httpbin.org/ip").json())Практический совет: Retry обеспечивает экспоненциальную задержку и учитывает заголовок Retry-After (в urllib3 v2.0+ включено по умолчанию, в v1.26 требуется явно задать для совместимости). Если в старом проекте всё ещё используется urllib3 1.26.x, оставьте строку respect_retry_after_header=True.
Переменные окружения: session.trust_env = False предотвращает вмешательство переменных окружения HTTP_PROXY/HTTPS_PROXY в явно заданную вами конфигурацию прокси.
⚠️ Безопасность POST: мы исключили POST из повторов по умолчанию. Если вам всё же необходимо повторять POST-запросы, используйте ключи идемпотентности и принимайте риск дублирования операций.
Разница между socks5 и socks5h (и когда можно угодить в ловушку)
socks5://...
Локальное разрешение DNS (ваша машина сначала определяет имя хоста, а затем поручает прокси подключиться к этому IP).
socks5h://...
Удалённое разрешение DNS (разрешается на прокси, предотвращает утечку DNS; для .onion и некоторых провайдеров обязательно). Для работы с даркнетом/.onion или когда провайдер требует удалённого разрешения, всегда используйте именно этот вариант записи.
Некоторые прокси-сети даже требуют удалённого разрешения DNS (цели заданы только именем хоста), поэтому используйте socks5h.
Ротация IP: запуск по требованию и режим циклического перебора
Провайдеры мобильных прокси обычно предлагают два способа ротации: смену IP по запросу через API или встроенный шлюз ротации.
Вариант 1: ротация по запросу через API (ниже в качестве примера используется Coronium.io; для других провайдеров подставьте их собственный интерфейс)
Многие провайдеры дают вам конечную точку перезапуска/ротации. Вызывайте её перед каждой партией сбора данных, чтобы получить новый IP:
import os
import requests
from urllib.parse import quote
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
# Your modem restart token (copy from dashboard.coronium.io)
ROTATION_TOKEN = os.getenv("ROTATION_TOKEN", "19a18981f48fa3907f04c19bdcba8666")
ROTATION_URL = f"https://dashboard.coronium.io/api/v1/modems/restart-by-token/{ROTATION_TOKEN}/"
# Proxy credentials
user = os.getenv("PROXY_USER")
pw = os.getenv("PROXY_PASS", "")
host = os.getenv("PROXY_HOST", "127.0.0.1")
port = int(os.getenv("PROXY_PORT", "1080"))
PROXY = f"socks5h://{quote(user)}:{quote(pw)}@{host}:{port}" if user else f"socks5h://{host}:{port}"
session = requests.Session()
session.trust_env = False
session.proxies.update({"http": PROXY, "https": PROXY})
retries = Retry(
total=5,
backoff_factor=0.5,
status_forcelist=[429, 500, 502, 503, 504],
allowed_methods=frozenset(["HEAD", "GET", "OPTIONS"]),
respect_retry_after_header=True,
)
adapter = HTTPAdapter(max_retries=retries, pool_connections=50, pool_maxsize=50)
session.mount("http://", adapter)
session.mount("https://", adapter)
def rotate_ip():
"""Trigger modem restart to get a new IP (actual wait time depends on your provider, usually 30-60 seconds)."""
try:
resp = requests.get(ROTATION_URL, timeout=10)
resp.raise_for_status()
print("✓ IP rotation triggered")
return True
except Exception as e:
print(f"⚠ Rotation failed: {e}")
return False
def fetch(url):
r = session.get(url, timeout=(5, 30))
r.raise_for_status()
return r
if __name__ == "__main__":
# Rotate before starting
rotate_ip()
# Wait for modem to reconnect (adjust timing based on your provider)
import time
time.sleep(45)
# Now scrape with fresh IP
print(fetch("https://httpbin.org/ip").json())Совет: вы можете напрямую скопировать свою ссылку перезапуска с dashboard.coronium.io - она содержит ваш уникальный token. Вызывайте её, когда нужен новый IP (например, каждые N запросов или после срабатывания ограничения скорости).
Вариант 2: циклический перебор нескольких прокси
Если у вас несколько прокси, используйте их по очереди с помощью itertools.cycle:
import itertools
from urllib.parse import quote
# List of proxies
PROXY_LIST = [
"socks5h://user1:pass1@host1:1080",
"socks5h://user2:pass2@host2:1080",
"socks5h://user3:pass3@host3:1080",
]
proxy_pool = itertools.cycle(PROXY_LIST)
def fetch_with_rotation(url):
proxy = next(proxy_pool)
proxies = {"http": proxy, "https": proxy}
r = requests.get(url, proxies=proxies, timeout=(5, 30))
r.raise_for_status()
return r
# Each call uses the next proxy in rotation
for i in range(5):
print(f"Request {i+1}:", fetch_with_rotation("https://httpbin.org/ip").json())⚠️ Внимание: циклический перебор подходит, когда у вас есть несколько независимых друг от друга прокси. Если же вы используете мобильный прокси типа 'один пул IP, где смена IP происходит по одному запросу', применяйте описанный выше вариант с запуском по требованию и не перехватывайте одну и ту же линию туда-сюда.
Справочные материалы и дополнительное чтение
Нужны SOCKS5/мобильные прокси, на которых приведённый выше код будет стабильно работать?
У нас есть узлы с реальными операторскими IP - идеальный вариант для автоматизации на Python.