ProxyScrape Python-Tutorial: Erstellen Sie einen asynchronen rotierenden Proxy-Pool mit httpx und SOCKS5

Überblick

Dies ist ein fortgeschrittenes Python-Tutorial. Anstatt eine Anfrage durch einen Proxy zu senden, bauen Sie einen kleinen Fetcher, der über die ProxyScrape-Endpunkte in Ihrem Plan rotiert, die Parallelität begrenzt, mit Jitter-Backoff wiederholt und fehlgeschlagene Endpunkte in eine Abklingzeit versetzt.

Sie werden:

• Installieren Sie httpx mit SOCKS5-Unterstützung.
• Speichern Sie ProxyScrape-Anmeldeinformationen in Umgebungsvariablen.
• Modellieren Sie Ihre HTTP- und SOCKS5-Endpunkte als Pool.
• Rotieren, drosseln, wiederholen und den Zustand pro Endpunkt melden.
• Führen Sie einen Smoke-Test durch, bevor Sie skalieren.

ProxyScrape bietet Residential-, Datacenter- und Mobile-Proxys, unterstützt HTTP und SOCKS5 und rechnet entweder pay-as-you-go oder pauschal monatlich ab. Was Sie wählen, ändert Kosten und Latenz, nicht den Client-Code – der folgende Code behandelt Endpunkte als Konfiguration.

Wählen Sie einen Proxy-Typ, bevor Sie Code schreiben

Voraussetzungen

• Python 3.10 oder neuer.
• Ein ProxyScrape-Plan mit Anmeldeinformationen sowie Host und Port für http und/oder socks5.
• Ausgehender Zugriff von Ihrem Computer auf den ProxyScrape-Host und -Port.

Schritte

Schritt 1 – Erstellen Sie ein Projekt und installieren Sie Abhängigkeiten

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

Das Extra [socks] installiert socksio. Ohne es löst httpx einen Fehler aus, wenn Sie zum ersten Mal einen socks5://-Endpunkt konfigurieren. httpx 0.26 und später verwenden das Argument proxy= (ältere Versionen verwendeten proxies=), also überprüfen Sie Ihre Version mit pip show httpx, wenn die Beispiele nicht übereinstimmen.

Schritt 2 – Speichern Sie Anmeldeinformationen und Endpunkte in Umgebungsvariablen

# 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")

Exportieren Sie die Werte in Ihrer Shell oder fügen Sie sie in CI aus einem Secrets-Manager ein:

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"

Prozentkodierung via quote() ist wichtig: Passwörter mit @, : oder / brechen sonst die URL-Analyse. Committen Sie diese Werte niemals.

Schritt 3 – Überprüfen Sie jeden Endpunkt vor der Skalierung

Führen Sie diese Checkliste in der Reihenfolge aus. Wenn ein Schritt fehlschlägt, beheben Sie ihn, bevor Sie den asynchronen Code anfassen.

1. Überprüfen Sie den HTTP/HTTPS-Endpunkt.
2. Überprüfen Sie den SOCKS5-Endpunkt.
3. Bestätigen Sie, dass die zurückgegebene IP nicht Ihre eigene IP ist.

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:// weist curl an, den Proxy DNS auflösen zu lassen. Wenn Ihr Python-Client sowohl socks5:// als auch socks5h:// anbietet, bevorzugen Sie die Variante mit h, wenn die lokale DNS-Auflösung fehlschlägt oder wenn Sie möchten, dass DNS zusammen mit der Anfrage Ihren Computer verlässt.

Schritt 4 – Erstellen Sie den rotierenden, nebenläufigen Fetcher

# 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()))

Design-Hinweise, die es wert sind, beibehalten zu werden:

• Ein AsyncClient pro Endpunkt. In httpx ist der Proxy an den Client gebunden, nicht an eine einzelne Anfrage, daher ist ein Client pro Endpunkt das einfachste korrekte Rotationsmodell.
• Der Semaphor begrenzt gleichzeitige Anfragen, egal wie viele URLs Sie übermitteln.
• 407 und 429 gesellen sich zu 5xx als wiederholbare Statuscodes und zählen als Endpunktfehler, sodass ein schlechter Endpunkt für eine Weile keine Anfragen mehr erhält.
• Fehlerstrafen wachsen linear und werden bei Erfolg zurückgesetzt, was verhindert, dass ein instabiler Endpunkt in einer engen Schleife wiederholt wird.

Schritt 5 – Smoke-Test und Optimierung der Parallelität

# 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())

Dann optimieren:

1. Führen Sie es einmal aus. Wenn failed 0 ist, erhöhen Sie concurrency um 4 und führen Sie es erneut aus.
2. Hören Sie auf zu erhöhen, wenn Latenz oder Timeouts zu steigen beginnen – das ist Ihre praktische Obergrenze für den gekauften Plan.
3. Wenn Fehler sich auf einen einzelnen Endpunkt konzentrieren, behandeln Sie diesen Endpunkt als ungesund und kontaktieren Sie den Support, anstatt härter zu wiederholen.

Plan-Hinweis: Bei pay-as-you-go-Abrechnung können Sie die Parallelität für einen Burst erhöhen; ein pauschal monatlicher Plan eignet sich normalerweise besser für gleichmäßige, geplante Jobs.

Fehlerbehebung

Zusammenfassung

• ProxyScrape unterstützt HTTP und SOCKS5; stellen Sie jeden Endpunkt aus Ihrem Plan als ProxyEndpoint dar und rotieren Sie über sie.
• Binden Sie einen Proxy an einen httpx.AsyncClient, begrenzen Sie gleichzeitige Arbeit mit einem Semaphor und wiederholen Sie mit Jitter-Backoff.
• Behandeln Sie 407-, 429- und 5xx-Antworten als Endpunktfehler statt als Benutzerfehler und kühlen Sie den Endpunkt ab.
• Überprüfen Sie die Endpunkte mit curl vor der Skalierung und erhöhen Sie dann die Parallelität schrittweise, bis Latenz – nicht Glück – Ihre Grenze wird.