US-Proxys in Python nutzen: requests, httpx, Scrapy und Playwright

Wie verwende ich einen US-Proxy mit Python requests?

Übergebe eine Proxy-URL für sowohl http als auch https im proxies Argument. Behalte das http:// Schema in der Proxy-URL auch für HTTPS-Ziele bei: Das Gateway öffnet einen HTTPS-Tunnel mit CONNECT. Für SOCKS5 installiere requests[socks] und nutze socks5h://, damit DNS-Abfragen ebenfalls auf der US-Seite erfolgen.

import requests

HOST = "gw.statesideip.com"
USER = "USERNAME-country-us-state-ny"  # rotating New York State IP
PASSWORD = "PASSWORD"

proxy = f"http://{USER}:{PASSWORD}@{HOST}:7777"
r = requests.get("https://api.ipify.org?format=json",
                 proxies={"http": proxy, "https": proxy}, timeout=30)
print(r.json())

# SOCKS5 (pip install "requests[socks]"); socks5h resolves DNS through the proxy
socks = f"socks5h://{USER}:{PASSWORD}@{HOST}:7778"
r = requests.get("https://api.ipify.org?format=json",
                 proxies={"http": socks, "https": socks}, timeout=30)
print(r.json())

Soll ich IPs pro Anfrage rotieren oder eine Sticky Session verwenden?

Rotieren Sie pro Anfrage für unabhängige Seitenabrufe und verwenden Sie eine Sticky Session für Aufgaben, die eine feste IP benötigen. Ein Benutzername ohne session Segment erhält bei jeder Anfrage eine neue IP. Das Hinzufügen von session-{id} und ttl-{minutes} behält eine IP für 1 bis 120 Minuten, pro Session mit dem Parameter ttl festgelegt.

• Rotierend: Suchergebnisprüfungen oder Katalogpreisprüfungen verteilt auf viele US Residential-Proxys.
• Sticky: Ein mehrseitiger Ablauf auf Ihrer eigenen Seite, z. B. ein Warenkorb-zu-Kasse QA-Test.

Rotierende Gateways weisen die Ausgangs-IP normalerweise pro Verbindung zu. Wenn mehrere Anfragen eine IP teilen, verwendet Ihr Client eine Verbindung erneut; ein neuer requests.get() Aufruf ohne Session öffnet eine neue Verbindung.

import uuid

def proxy_url(state=None, city=None, zip_code=None, session=None, ttl=10):
    parts = ["USERNAME", "country-us"]
    if state:
        parts.append(f"state-{state}")
    if city:
        parts.append(f"city-{city}")
    if zip_code:
        parts.append(f"zip-{zip_code}")
    if session:
        parts += [f"session-{session}", f"ttl-{ttl}"]
    user = "-".join(parts)
    return f"http://{user}:[email protected]:7777"

rotating = proxy_url(state="tx", city="houston")  # new IP per request
sticky = proxy_url(zip_code="10001", session=uuid.uuid4().hex[:8], ttl=30)  # one IP, 30 min

Wie sende ich viele Anfragen gleichzeitig mit httpx async?

Verwenden Sie einen httpx.AsyncClient mit dem Argument proxy und begrenzen Sie die gleichzeitig laufenden Anfragen mit einem asyncio.Semaphore. Beginnen Sie mit etwa 5 Anfragen gleichzeitig pro Zielseite. Erhöhen Sie das Limit nur, solange die Fehlerquote niedrig bleibt.

Das folgende Beispiel deaktiviert keep-alive, sodass jede Anfrage eine neue Verbindung öffnet und eine neue IP erhalten kann. Entfernen Sie diese Zeile, wenn Sie eine Sticky Session verwenden. Neuere httpx-Versionen akzeptieren proxy=; das ältere Argument proxies= wurde in httpx 0.28 entfernt.

import asyncio
import httpx

PROXY = "http://USERNAME-country-us-state-ca:[email protected]:7777"
URLS = ["https://api.ipify.org?format=json"] * 20

async def fetch(client, sem, url):
    async with sem:  # caps requests in flight
        r = await client.get(url)
        r.raise_for_status()
        return r.json()

async def main():
    sem = asyncio.Semaphore(5)
    limits = httpx.Limits(max_keepalive_connections=0)  # new connection, new IP
    async with httpx.AsyncClient(proxy=PROXY, limits=limits, timeout=30) as client:
        results = await asyncio.gather(*(fetch(client, sem, u) for u in URLS),
                                       return_exceptions=True)
    for res in results:
        print(res)

asyncio.run(main())

Wie gehe ich mit Wiederholungen, Backoff und Rate Limits um?

Wiederholen Sie nur temporäre Fehler, warten Sie nach jedem Versuch länger und fügen Sie zufälliges Jitter hinzu. Wiederholen bei Timeouts, Verbindungsfehlern und Statuscodes 429, 500, 502, 503 und 504. Beachten Sie einen Retry-After Header, wenn die Seite einen sendet. Wiederholen Sie niemals einen 407: das bedeutet, dass Benutzername oder Passwort falsch sind.

Respektieren Sie die robots.txt und Nutzungsbedingungen jeder Seite und halten Sie die Anfragerate nahe am normalen Browsing. StatesideIP blockiert Banking-, Regierungs- und Zahlungsportale sowie SMTP-Ports gemäß seiner Acceptable Use Policy.

import random
import time
import requests

RETRY_STATUS = {429, 500, 502, 503, 504}

def get_with_retries(url, proxies, attempts=5):
    for attempt in range(attempts):
        try:
            r = requests.get(url, proxies=proxies, timeout=30)
            if r.status_code == 407:
                raise RuntimeError("407: check USERNAME and PASSWORD")  # do not retry
            if r.status_code not in RETRY_STATUS:
                return r
            ra = r.headers.get("Retry-After", "")
            wait = int(ra) if ra.isdigit() else 2 ** attempt
        except (requests.ConnectionError, requests.Timeout):
            wait = 2 ** attempt
        time.sleep(wait + random.uniform(0, 1))  # backoff plus jitter
    raise RuntimeError(f"gave up on {url}")

Wie verwende ich US Residential-Proxys in Scrapy?

Setzen Sie den Proxy pro Anfrage in meta["proxy"], mit Benutzername und Passwort in der URL. Scrapy's integrierte Proxy-Middleware wandelt diese in einen Proxy-Authorization Header um. Behalten Sie ROBOTSTXT_OBEY = True bei und lassen Sie AutoThrottle den Crawl verlangsamen, wenn eine Seite langsam reagiert.

# settings.py
ROBOTSTXT_OBEY = True
CONCURRENT_REQUESTS_PER_DOMAIN = 4
AUTOTHROTTLE_ENABLED = True
RETRY_TIMES = 3
RETRY_HTTP_CODES = [429, 500, 502, 503, 504]

# spiders/ip_check.py
import scrapy

PROXY = "http://USERNAME-country-us-state-fl-city-miami:[email protected]:7777"

class IpCheckSpider(scrapy.Spider):
    name = "ip_check"
    start_urls = ["https://api.ipify.org?format=json"]

    def start_requests(self):
        for url in self.start_urls:
            yield scrapy.Request(url, meta={"proxy": PROXY}, dont_filter=True)

    def parse(self, response):
        yield {"ip": response.json()["ip"]}

Wie verwende ich US Residential-Proxys mit Playwright?

Übergeben Sie ein proxy Dictionary mit Server, Benutzername und Passwort beim Starten des Browsers. Ein Browser lädt viele Dateien pro Seite, daher verwenden Sie eine Sticky Session: eine Seitenansicht behält eine US Residential Proxy IP. Verwenden Sie für jede neue Aufgabe eine neue Session-ID.

import asyncio
from playwright.async_api import async_playwright

async def main():
    async with async_playwright() as p:
        browser = await p.chromium.launch(proxy={
            "server": "http://gw.statesideip.com:7777",
            "username": "USERNAME-country-us-state-il-city-chicago-session-pw1-ttl-15",
            "password": "PASSWORD",
        })
        page = await browser.new_page()
        await page.goto("https://api.ipify.org?format=json")
        print(await page.text_content("body"))
        await browser.close()

asyncio.run(main())

Warum schlägt meine Python Proxy-Anfrage fehl?

Die meisten Fehler entstehen durch falsche Zugangsdaten, zu enge Zielauswahl oder einen Client, der Verbindungen wiederverwendet.

• 407 Proxy Authentication Required: Benutzername oder Passwort ist falsch oder ein Segment ist falsch geschrieben. Vergleichen Sie es mit USERNAME-country-{cc}-state-{state}-city-{city}-zip-{zip}-asn-{asn}-session-{id}-ttl-{minutes}.
• Timeouts bei ZIP- oder ASN-Zielen: Ein zu enger Filter nutzt weniger Residential-Proxys. Erweitern Sie auf Stadt oder Bundesstaat oder erhöhen Sie das Timeout.
• Immer dieselbe IP: Ihr Client hält die Verbindung offen. Deaktivieren Sie keep-alive oder entfernen Sie das session Segment.
• SSL-Fehler: Verwenden Sie http:// für die Proxy-URL, nicht https://.

Der Getting-Started Guide behandelt curl und Node.