Proxys in Python mit aiohttp verwenden

Proxys in Python mit aiohttp verwenden

aiohttp und Proxys

aiohttp ist die beliebteste asynchrone HTTP-Bibliothek für Python. Sie ermöglicht das Senden von Tausenden gleichzeitiger Anfragen, was sie ideal für Parsing und Automatisierung macht. Die Proxy-Unterstützung in aiohttp ist ein Schlüsselelement für Aufgaben im großen Maßstab.

Grundlegende Proxy-Nutzung

HTTP-Proxys

Die einfachste Methode, einen HTTP-Proxy in aiohttp zu verwenden, ist mit dem proxy-Parameter in der Request-Methode:

import aiohttp
import asyncio

async def fetch_with_proxy():
    proxy = "http://proxy_ip:port"
    async with aiohttp.ClientSession() as session:
        async with session.get(
            "https://httpbin.org/ip",
            proxy=proxy
        ) as response:
            data = await response.json()
            print(data)

asyncio.run(fetch_with_proxy())

Proxys mit Authentifizierung

Für Proxys mit Benutzername und Passwort verwenden Sie BasicAuth:

import aiohttp
import asyncio

async def fetch_with_auth_proxy():
    proxy = "http://proxy_ip:port"
    proxy_auth = aiohttp.BasicAuth("username", "password")

    async with aiohttp.ClientSession() as session:
        async with session.get(
            "https://httpbin.org/ip",
            proxy=proxy,
            proxy_auth=proxy_auth
        ) as response:
            data = await response.json()
            print(data)

asyncio.run(fetch_with_auth_proxy())

Ein alternatives Format sind Anmeldeinformationen in der URL:

proxy = "http://username:password@proxy_ip:port"

SOCKS5-Proxys

aiohttp-socks installieren

aiohttp unterstützt SOCKS nicht nativ. Installieren Sie aiohttp-socks:

pip install aiohttp-socks

Verwendung

import aiohttp
from aiohttp_socks import ProxyConnector

async def fetch_with_socks():
    connector = ProxyConnector.from_url("socks5://user:pass@proxy_ip:port")

    async with aiohttp.ClientSession(connector=connector) as session:
        async with session.get("https://httpbin.org/ip") as response:
            data = await response.json()
            print(data)

asyncio.run(fetch_with_socks())

Unterstützte Protokolle

ProxyConnector unterstützt:
- socks5://
- socks4://
- http://
- https://

Proxy-Rotation

Einfache Rotation

import aiohttp
import asyncio
import random

PROXIES = [
    "http://user:pass@proxy1:port",
    "http://user:pass@proxy2:port",
    "http://user:pass@proxy3:port",
]

async def fetch_with_rotation(url):
    proxy = random.choice(PROXIES)
    async with aiohttp.ClientSession() as session:
        async with session.get(url, proxy=proxy) as response:
            return await response.text()

async def main():
    urls = ["https://example.com"] * 10
    tasks = [fetch_with_rotation(url) for url in urls]
    results = await asyncio.gather(*tasks, return_exceptions=True)
    for r in results:
        if isinstance(r, Exception):
            print(f"Error: {r}")
        else:
            print(f"OK: {len(r)} bytes")

asyncio.run(main())

Rotation mit Ausschluss fehlgeschlagener Proxys

import aiohttp
import asyncio
from collections import deque

class ProxyRotator:
    def __init__(self, proxies):
        self.proxies = deque(proxies)
        self.failed = set()

    def get_proxy(self):
        for _ in range(len(self.proxies)):
            proxy = self.proxies[0]
            self.proxies.rotate(-1)
            if proxy not in self.failed:
                return proxy
        raise Exception("All proxies failed")

    def mark_failed(self, proxy):
        self.failed.add(proxy)

    def mark_success(self, proxy):
        self.failed.discard(proxy)

async def fetch(session, url, rotator, retries=3):
    for attempt in range(retries):
        proxy = rotator.get_proxy()
        try:
            async with session.get(url, proxy=proxy, timeout=aiohttp.ClientTimeout(total=10)) as resp:
                if resp.status == 200:
                    rotator.mark_success(proxy)
                    return await resp.text()
                elif resp.status == 403:
                    rotator.mark_failed(proxy)
        except Exception:
            rotator.mark_failed(proxy)
    return None

Gleichzeitige Anfragen mit Limits

Semaphore zur Parallelitätskontrolle

import aiohttp
import asyncio

async def fetch(session, url, proxy, semaphore):
    async with semaphore:
        try:
            async with session.get(url, proxy=proxy, timeout=aiohttp.ClientTimeout(total=15)) as resp:
                return await resp.text()
        except Exception as e:
            return None

async def main():
    urls = [f"https://example.com/page/{i}" for i in range(100)]
    proxy = "http://user:pass@proxy:port"
    semaphore = asyncio.Semaphore(10)  # max 10 concurrent

    async with aiohttp.ClientSession() as session:
        tasks = [fetch(session, url, proxy, semaphore) for url in urls]
        results = await asyncio.gather(*tasks)
        success = sum(1 for r in results if r)
        print(f"Success: {success}/{len(urls)}")

asyncio.run(main())

Session-Konfiguration

Timeout

timeout = aiohttp.ClientTimeout(
    total=30,        # Gesamt-Timeout
    connect=10,      # Verbindungs-Timeout
    sock_read=10     # Socket-Lese-Timeout
)
session = aiohttp.ClientSession(timeout=timeout)

Header

headers = {
    "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",
    "Accept": "text/html,application/xhtml+xml",
    "Accept-Language": "en-US,en;q=0.9",
}
session = aiohttp.ClientSession(headers=headers)

SSL

# SSL-Verifizierung deaktivieren (zum Testen)
import ssl
ssl_context = ssl.create_default_context()
ssl_context.check_hostname = False
ssl_context.verify_mode = ssl.CERT_NONE

session = aiohttp.ClientSession(connector=aiohttp.TCPConnector(ssl=ssl_context))

Fehlerbehandlung

import aiohttp

async def safe_fetch(session, url, proxy):
    try:
        async with session.get(url, proxy=proxy) as resp:
            if resp.status == 200:
                return await resp.text()
            elif resp.status == 407:
                print("Proxy-Authentifizierung erforderlich")
            elif resp.status == 403:
                print("Verboten - Proxy könnte gesperrt sein")
            elif resp.status == 429:
                print("Ratenbegrenzung erreicht - verlangsamen Sie")
            return None
    except aiohttp.ClientProxyConnectionError:
        print("Verbindung zum Proxy nicht möglich")
    except aiohttp.ClientConnectorError:
        print("Verbindungsfehler")
    except asyncio.TimeoutError:
        print("Anfrage-Timeout")
    except Exception as e:
        print(f"Unerwarteter Fehler: {e}")
    return None

aiohttp vs. requests

Fazit

aiohttp ist die beste Wahl für die asynchrone Proxy-Nutzung in Python. Die Unterstützung für HTTP und SOCKS5 (via aiohttp-socks), Proxy-Rotation, Parallelitätskontrolle und Fehlerbehandlung machen es zu einem leistungsstarken Werkzeug für groß angelegte Parsing- und Automatisierungsaufgaben.