Tutoriel ProxyScrape Python : Créez un pool de proxies rotatifs asynchrones avec httpx et SOCKS5

Aperçu

Ceci est un tutoriel Python avancé. Au lieu d'envoyer une requête via un seul proxy, vous construisez un petit récupérateur qui alterne entre les points de terminaison ProxyScrape de votre forfait, plafonne la concurrence, réessaie avec un backoff aléatoire et met en veille les points de terminaison défaillants dans un délai de récupération.

Vous allez :

• Installer httpx avec le support SOCKS5.
• Conserver les identifiants ProxyScrape dans des variables d'environnement.
• Modéliser vos points de terminaison HTTP et SOCKS5 comme un pool.
• Alterner, limiter, réessayer et signaler l'état de santé de chaque point de terminaison.
• Effectuer un test de fumée avant de monter en charge.

ProxyScrape propose des proxies résidentiels, de datacenter et mobiles, prend en charge HTTP et SOCKS5, et facture soit à l'usage, soit au forfait mensuel. Votre choix influence le coût et la latence, pas le code client — le code ci-dessous traite les points de terminaison comme de la configuration.

Choisissez un type de proxy avant d'écrire du code

Prérequis

• Python 3.10 ou plus récent.
• Un forfait ProxyScrape avec identifiants, ainsi que l'hôte et le port pour http et/ou socks5.
• Accès sortant depuis votre machine vers l'hôte et le port ProxyScrape.

Étapes

Étape 1 — Créez un projet et installez les dépendances

mkdir proxyscrape-async
cd proxyscrape-async
python -m venv .venv
source .venv/bin/activate
pip install "httpx[socks]"

L'extra [socks] installe socksio. Sans cela, httpx lève une erreur la première fois que vous configurez un point de terminaison socks5://. httpx 0.26 et versions ultérieures utilisent l'argument proxy= (les versions plus anciennes utilisaient proxies=), donc vérifiez votre version avec pip show httpx si les exemples ne correspondent pas.

Étape 2 — Conservez les identifiants et les points de terminaison dans des variables d'environnement

# pool.py
import os
from dataclasses import dataclass
from urllib.parse import quote


@dataclass(frozen=True)
class ProxyEndpoint:
    label: str
    url: str


def endpoint_url(scheme: str, host: str, port: str, username: str, password: str) -> str:
    user = quote(username, safe="")
    secret = quote(password, safe="")
    return f"{scheme}://{user}:{secret}@{host}:{port}"


HOST = os.environ["PROXYSCRAPE_HOST"]
USERNAME = os.environ["PROXYSCRAPE_USERNAME"]
PASSWORD = os.environ["PROXYSCRAPE_PASSWORD"]

POOL: list[ProxyEndpoint] = []

if http_port := os.environ.get("PROXYSCRAPE_HTTP_PORT"):
    POOL.append(ProxyEndpoint("http-1", endpoint_url("http", HOST, http_port, USERNAME, PASSWORD)))

if socks_port := os.environ.get("PROXYSCRAPE_SOCKS5_PORT"):
    POOL.append(ProxyEndpoint("socks5-1", endpoint_url("socks5", HOST, socks_port, USERNAME, PASSWORD)))

if not POOL:
    raise SystemExit("Set PROXYSCRAPE_HTTP_PORT and/or PROXYSCRAPE_SOCKS5_PORT")

Exportez les valeurs dans votre shell, ou injectez-les depuis un gestionnaire de secrets dans la CI :

export PROXYSCRAPE_HOST="host-from-your-plan"
export PROXYSCRAPE_USERNAME="your-username"
export PROXYSCRAPE_PASSWORD="your-password"
export PROXYSCRAPE_HTTP_PORT="port-from-your-plan"
export PROXYSCRAPE_SOCKS5_PORT="port-from-your-plan"

L'encodage en pourcentage via quote() est important : les mots de passe contenant @, :, ou / cassent sinon l'analyse d'URL. Ne validez jamais ces valeurs dans le dépôt.

Étape 3 — Vérifiez chaque point de terminaison avant de monter en charge

Exécutez cette liste de contrôle dans l'ordre. Si une étape échoue, corrigez-la avant de toucher au code asynchrone.

1. Vérifiez le point de terminaison HTTP/HTTPS.
2. Vérifiez le point de terminaison SOCKS5.
3. Confirmez que l'IP renvoyée n'est pas votre propre IP.

curl -sS -x "http://$PROXYSCRAPE_USERNAME:$PROXYSCRAPE_PASSWORD@$PROXYSCRAPE_HOST:$PROXYSCRAPE_HTTP_PORT" "https://api.ipify.org?format=json"
curl -sS -x "socks5h://$PROXYSCRAPE_USERNAME:$PROXYSCRAPE_PASSWORD@$PROXYSCRAPE_HOST:$PROXYSCRAPE_SOCKS5_PORT" "https://api.ipify.org?format=json"

socks5h:// indique à curl de laisser le proxy résoudre le DNS. Si votre client Python propose à la fois socks5:// et socks5h://, préférez la variante h lorsque la résolution DNS locale échoue ou lorsque vous voulez que le DNS quitte votre machine avec la requête.

Étape 4 — Construisez le récupérateur rotatif et concurrent

# fetch_async.py
from __future__ import annotations

import asyncio
import itertools
import random
import time

import httpx

from pool import POOL, ProxyEndpoint


class RotatingFetcher:
    """Round-robin fetcher with concurrency limits, backoff, and cooldowns."""

    def __init__(
        self,
        pool: list[ProxyEndpoint],
        concurrency: int = 8,
        max_attempts: int = 4,
        timeout: float = 20.0,
        cooldown_seconds: float = 45.0,
    ) -> None:
        if not pool:
            raise ValueError("Proxy pool is empty")
        self._pool = pool
        self._cycle = itertools.cycle(pool)
        self._semaphore = asyncio.Semaphore(concurrency)
        self._max_attempts = max_attempts
        self._cooldown_seconds = cooldown_seconds
        self._failures: dict[str, int] = {}
        self._cooldown_until: dict[str, float] = {}
        self._clients = {
            item.label: httpx.AsyncClient(
                proxy=item.url,
                timeout=timeout,
                follow_redirects=True,
                headers={"User-Agent": "Mozilla/5.0 (compatible; ProxyScoutTutorial/1.0)"},
            )
            for item in pool
        }

    def _pick(self) -> ProxyEndpoint:
        now = time.monotonic()
        for _ in range(len(self._pool) * 2):
            candidate = next(self._cycle)
            if self._cooldown_until.get(candidate.label, 0.0) <= now:
                return candidate
        return min(self._pool, key=lambda item: self._cooldown_until.get(item.label, 0.0))

    def _record_failure(self, label: str) -> None:
        self._failures[label] = self._failures.get(label, 0) + 1
        penalty = self._cooldown_seconds * self._failures[label]
        self._cooldown_until[label] = time.monotonic() + penalty

    def _record_success(self, label: str) -> None:
        self._failures[label] = 0
        self._cooldown_until[label] = 0.0

    @staticmethod
    def _backoff(attempt: int) -> float:
        return min((2 ** attempt) + random.uniform(0, 0.5), 30.0)

    async def fetch(self, url: str) -> httpx.Response:
        last_error: Exception | None = None
        for attempt in range(1, self._max_attempts + 1):
            item = self._pick()
            try:
                async with self._semaphore:
                    response = await self._clients[item.label].get(url)
                if response.status_code in {407, 429} or response.status_code >= 500:
                    raise httpx.HTTPStatusError(
                        f"retryable status {response.status_code}",
                        request=response.request,
                        response=response,
                    )
                self._record_success(item.label)
                return response
            except Exception as exc:
                last_error = exc
                self._record_failure(item.label)
                if attempt < self._max_attempts:
                    await asyncio.sleep(self._backoff(attempt))
        raise RuntimeError(f"All {self._max_attempts} attempts failed for {url}") from last_error

    def report(self) -> str:
        active = {label: count for label, count in self._failures.items() if count}
        if not active:
            return "no endpoints have failed yet"
        return ", ".join(f"{label}={count}" for label, count in sorted(active.items()))

    async def aclose(self) -> None:
        await asyncio.gather(*(client.aclose() for client in self._clients.values()))

Notes de conception à conserver :

• Un AsyncClient par point de terminaison. Dans httpx, le proxy est lié au client, et non à une requête individuelle, donc un client par point de terminaison est le modèle de rotation correct le plus simple.
• Le sémaphore plafonne les requêtes en cours, quel que soit le nombre d'URL que vous soumettez.
• 407 et 429 rejoignent 5xx comme statuts réessayables, et ils comptent comme des échecs de point de terminaison, de sorte qu'un mauvais point de terminaison cesse de recevoir du trafic pendant un moment.
• Les pénalités d'échec augmentent linéairement et se réinitialisent en cas de succès, ce qui empêche un point de terminaison instable d'être réessayé en boucle serrée.

Étape 5 — Test de fumée et réglage de la concurrence

# run_smoke_test.py
import asyncio

import httpx

from fetch_async import RotatingFetcher
from pool import POOL


async def main() -> None:
    fetcher = RotatingFetcher(pool=POOL, concurrency=8)
    urls = [f"https://example.com/?page={index}" for index in range(1, 21)]
    try:
        results = await asyncio.gather(*(fetcher.fetch(url) for url in urls), return_exceptions=True)
        ok = sum(1 for item in results if isinstance(item, httpx.Response))
        print(f"success={ok} failed={len(results) - ok}")
        print(f"endpoint health: {fetcher.report()}")
    finally:
        await fetcher.aclose()


if __name__ == "__main__":
    asyncio.run(main())

Ensuite, ajustez :

1. Exécutez-le une fois. Si failed est 0, augmentez concurrency de 4 et exécutez à nouveau.
2. Arrêtez d'augmenter lorsque la latence ou les délais d'attente commencent à grimper — c'est votre plafond pratique pour le forfait que vous avez acheté.
3. Si les échecs se concentrent sur un seul point de terminaison, considérez ce point de terminaison comme défaillant et contactez le support au lieu de réessayer plus fort.

Note sur le forfait : avec la facturation à l'usage, vous pouvez augmenter la concurrence pour un pic ; un forfait mensuel fixe convient généralement mieux aux tâches régulières et planifiées.

Dépannage

Résumé

• ProxyScrape prend en charge HTTP et SOCKS5 ; représentez chaque point de terminaison de votre forfait comme un ProxyEndpoint et alternez entre eux.
• Liez un proxy à un httpx.AsyncClient, plafonnez le travail en cours avec un sémaphore, et réessayez avec un backoff aléatoire.
• Traitez les réponses 407, 429 et 5xx comme des échecs de point de terminaison plutôt que des erreurs utilisateur, et mettez le point de terminaison en cooldown.
• Vérifiez les points de terminaison avec curl avant de monter en charge, puis augmentez progressivement la concurrence jusqu'à ce que la latence — et non la chance — devienne votre limite.