Zurück zum Blog

Proxy-Einrichtung in Python Requests und aiohttp: umfassender Leitfaden mit Codebeispielen

Schritt-für-Schritt-Anleitung zur Integration von Proxys in Python requests und aiohttp mit Codebeispielen für synchrone und asynchrone Anfragen, IP-Rotation und Fehlerbehandlung.

📅13. Februar 2026
```html

Bei der Entwicklung von Parsern, der Automatisierung der Datensammlung oder dem Testen von Webdiensten ist es oft erforderlich, Proxys in Python zu verwenden. Die Bibliotheken requests und aiohttp bieten flexible Mechanismen zur Arbeit mit Proxys, aber ihre Einrichtung hat wichtige Nuancen. In diesem Handbuch werden wir synchrone und asynchrone Ansätze untersuchen, Beispiele für HTTP- und SOCKS5-Proxys zeigen, IP-Rotation und Fehlerbehandlung betrachten.

Grundlegende Proxy-Einrichtung in requests

Die Bibliothek requests ist der Standard für HTTP-Anfragen in Python. Die Proxy-Einrichtung erfolgt über den Parameter proxies, der ein Dictionary mit Protokollen und Adressen von Proxy-Servern akzeptiert.

Ein einfaches Beispiel mit HTTP-Proxy:

import requests

# Proxy-Einrichtung
proxies = {
    'http': 'http://123.45.67.89:8080',
    'https': 'http://123.45.67.89:8080'
}

# Anfrage über Proxy ausführen
response = requests.get('https://httpbin.org/ip', proxies=proxies)
print(response.json())  # {'origin': '123.45.67.89'}

Hinweis: Für HTTPS-Anfragen muss auch das Protokoll http:// im Wert des Proxys angegeben werden (nicht https://). Dies liegt daran, dass die Verbindung zum Proxy-Server über HTTP hergestellt wird und dann über die Methode CONNECT ein Tunnel für den HTTPS-Verkehr erstellt wird.

Verwendung von Umgebungsvariablen:

Die Bibliothek requests liest automatisch Proxys aus den Umgebungsvariablen HTTP_PROXY und HTTPS_PROXY:

import os
import requests

# Einrichtung über Umgebungsvariablen
os.environ['HTTP_PROXY'] = 'http://123.45.67.89:8080'
os.environ['HTTPS_PROXY'] = 'http://123.45.67.89:8080'

# Proxy wird automatisch angewendet
response = requests.get('https://httpbin.org/ip')
print(response.json())

Dieser Ansatz ist praktisch für Containerisierung (Docker) oder wenn Proxys auf Systemebene konfiguriert werden. Für Flexibilität wird jedoch empfohlen, den Parameter proxies explizit zu übergeben.

Authentifizierung und SOCKS5 in requests

Die meisten kommerziellen Proxy-Dienste erfordern eine Authentifizierung mit Benutzername und Passwort. In requests wird dies über das URL-Format mit Anmeldedaten realisiert.

HTTP-Proxy mit Authentifizierung:

import requests

# Format: http://username:password@host:port
proxies = {
    'http': 'http://user123:pass456@proxy.example.com:8080',
    'https': 'http://user123:pass456@proxy.example.com:8080'
}

response = requests.get('https://httpbin.org/ip', proxies=proxies)
print(response.json())

Einrichtung von SOCKS5-Proxy:

Für die Arbeit mit SOCKS5 ist eine zusätzliche Bibliothek requests[socks] oder PySocks erforderlich. Installation:

pip install requests[socks]

Beispiel für die Verwendung von SOCKS5:

import requests

# SOCKS5 ohne Authentifizierung
proxies = {
    'http': 'socks5://123.45.67.89:1080',
    'https': 'socks5://123.45.67.89:1080'
}

# SOCKS5 mit Authentifizierung
proxies_auth = {
    'http': 'socks5://user:pass@123.45.67.89:1080',
    'https': 'socks5://user:pass@123.45.67.89:1080'
}

response = requests.get('https://httpbin.org/ip', proxies=proxies_auth)
print(response.json())

SOCKS5-Proxys sind besonders nützlich bei der Arbeit mit residential Proxys, da dieses Protokoll ein zuverlässigeres Tunneling des Verkehrs ermöglicht und UDP unterstützt (notwendig für einige Anwendungen).

Proxy-Rotation in requests

Bei der Analyse großer Datenmengen führt die Verwendung einer einzigen IP-Adresse zu Sperrungen. Proxy-Rotation ist der zyklische Wechsel der IP-Adresse zur Lastverteilung und Umgehung von Rate-Limits.

Einfache Rotation über eine Liste:

import requests
import itertools

# Liste von Proxy-Servern
proxy_list = [
    'http://user:pass@proxy1.example.com:8080',
    'http://user:pass@proxy2.example.com:8080',
    'http://user:pass@proxy3.example.com:8080',
]

# Erstellen eines unendlichen Iterators
proxy_pool = itertools.cycle(proxy_list)

# Ausführen von Anfragen mit Rotation
for i in range(10):
    proxy = next(proxy_pool)
    proxies = {'http': proxy, 'https': proxy}
    
    try:
        response = requests.get('https://httpbin.org/ip', proxies=proxies, timeout=5)
        print(f"Anfrage {i+1}: IP = {response.json()['origin']}")
    except Exception as e:
        print(f"Fehler mit Proxy {proxy}: {e}")

Rotation mit Sessions zur Speicherung von Cookies:

import requests
from itertools import cycle

class ProxyRotator:
    def __init__(self, proxy_list):
        self.proxy_pool = cycle(proxy_list)
        self.session = requests.Session()
    
    def get(self, url, **kwargs):
        proxy = next(self.proxy_pool)
        self.session.proxies = {'http': proxy, 'https': proxy}
        return self.session.get(url, **kwargs)

# Verwendung
proxy_list = [
    'http://user:pass@proxy1.example.com:8080',
    'http://user:pass@proxy2.example.com:8080',
]

rotator = ProxyRotator(proxy_list)

for i in range(5):
    response = rotator.get('https://httpbin.org/ip', timeout=5)
    print(f"Anfrage {i+1}: {response.json()['origin']}")

Zufällige Rotation für Unvorhersehbarkeit:

import requests
import random

proxy_list = [
    'http://user:pass@proxy1.example.com:8080',
    'http://user:pass@proxy2.example.com:8080',
    'http://user:pass@proxy3.example.com:8080',
]

def get_random_proxy():
    proxy = random.choice(proxy_list)
    return {'http': proxy, 'https': proxy}

# Jede Anfrage mit zufälligem Proxy
for i in range(5):
    response = requests.get('https://httpbin.org/ip', proxies=get_random_proxy(), timeout=5)
    print(f"Anfrage {i+1}: {response.json()['origin']}")

Zufällige Rotation ist effektiver bei der Arbeit mit Websites, die Anfrage-Muster verfolgen. Eine sequenzielle Änderung der IP kann verdächtig erscheinen, während die zufällige Auswahl das Verhalten verschiedener Benutzer imitiert.

Proxy-Einrichtung in aiohttp

Die Bibliothek aiohttp ist für asynchrone HTTP-Anfragen konzipiert und entscheidend für stark belastete Parser. Die Proxy-Einrichtung unterscheidet sich von requests — es wird der Parameter proxy (im Singular) verwendet.

Grundlegendes Beispiel mit HTTP-Proxy:

import aiohttp
import asyncio

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

# Ausführung
asyncio.run(fetch_with_proxy())

Proxy mit Authentifizierung:

In aiohttp wird die Authentifizierung über das Objekt aiohttp.BasicAuth oder direkt in der URL übergeben:

import aiohttp
import asyncio

async def fetch_with_auth_proxy():
    # Variante 1: Anmeldedaten in der URL
    proxy = 'http://user123:pass456@proxy.example.com:8080'
    
    async with aiohttp.ClientSession() as session:
        async with session.get('https://httpbin.org/ip', proxy=proxy) as response:
            print(await response.json())

# Variante 2: Über BasicAuth (für einige Proxys)
async def fetch_with_basic_auth():
    proxy = 'http://proxy.example.com:8080'
    proxy_auth = aiohttp.BasicAuth('user123', 'pass456')
    
    async with aiohttp.ClientSession() as session:
        async with session.get('https://httpbin.org/ip', 
                                proxy=proxy, 
                                proxy_auth=proxy_auth) as response:
            print(await response.json())

asyncio.run(fetch_with_auth_proxy())

SOCKS5 in aiohttp:

Für SOCKS5 ist die Bibliothek aiohttp-socks erforderlich:

pip install aiohttp-socks
import asyncio
from aiohttp_socks import ProxyConnector
import aiohttp

async def fetch_with_socks5():
    connector = ProxyConnector.from_url('socks5://user:pass@123.45.67.89:1080')
    
    async with aiohttp.ClientSession(connector=connector) as session:
        async with session.get('https://httpbin.org/ip') as response:
            print(await response.json())

asyncio.run(fetch_with_socks5())

Bei der Arbeit mit mobilen Proxys für das Parsen von sozialen Netzwerken oder Marktplätzen wird empfohlen, genau aiohttp zu verwenden — die Asynchronität ermöglicht die Verarbeitung von Hunderten von Anfragen parallel, ohne den Ausführungsfluss zu blockieren.

Asynchrone Rotation und Proxy-Pool

Für stark belastete Parser ist eine effektive Proxy-Rotation mit Fehlerbehandlung und automatischem Austausch nicht funktionierender IPs entscheidend. Lassen Sie uns fortgeschrittene Muster für aiohttp betrachten.

Klasse zur Verwaltung des Proxy-Pools:

import aiohttp
import asyncio
from itertools import cycle
from typing import List, Optional

class ProxyPool:
    def __init__(self, proxy_list: List[str]):
        self.proxy_list = proxy_list
        self.proxy_cycle = cycle(proxy_list)
        self.failed_proxies = set()
    
    def get_next_proxy(self) -> Optional[str]:
        """Nächsten funktionierenden Proxy abrufen"""
        for _ in range(len(self.proxy_list)):
            proxy = next(self.proxy_cycle)
            if proxy not in self.failed_proxies:
                return proxy
        return None  # Alle Proxys sind nicht verfügbar
    
    def mark_failed(self, proxy: str):
        """Proxy als nicht funktionierend markieren"""
        self.failed_proxies.add(proxy)
        print(f"Proxy {proxy} als nicht verfügbar markiert")
    
    async def fetch(self, session: aiohttp.ClientSession, url: str, **kwargs):
        """Anfrage mit automatischem Wechsel des Proxys bei Fehlern ausführen"""
        max_retries = 3
        
        for attempt in range(max_retries):
            proxy = self.get_next_proxy()
            if not proxy:
                raise Exception("Alle Proxys sind nicht verfügbar")
            
            try:
                async with session.get(url, proxy=proxy, timeout=aiohttp.ClientTimeout(total=10), **kwargs) as response:
                    return await response.json()
            except (aiohttp.ClientError, asyncio.TimeoutError) as e:
                print(f"Fehler mit Proxy {proxy}: {e}")
                self.mark_failed(proxy)
                continue
        
        raise Exception(f"Anfrage konnte nach {max_retries} Versuchen nicht ausgeführt werden")

# Verwendung
async def main():
    proxy_list = [
        'http://user:pass@proxy1.example.com:8080',
        'http://user:pass@proxy2.example.com:8080',
        'http://user:pass@proxy3.example.com:8080',
    ]
    
    pool = ProxyPool(proxy_list)
    
    async with aiohttp.ClientSession() as session:
        # Ausführen von 10 Anfragen mit automatischer Rotation
        tasks = [pool.fetch(session, 'https://httpbin.org/ip') for _ in range(10)]
        results = await asyncio.gather(*tasks, return_exceptions=True)
        
        for i, result in enumerate(results):
            if isinstance(result, Exception):
                print(f"Anfrage {i+1} endete mit Fehler: {result}")
            else:
                print(f"Anfrage {i+1}: IP = {result.get('origin')}")

asyncio.run(main())

Parallele Verarbeitung mit Begrenzung der gleichzeitigen Anfragen:

import aiohttp
import asyncio
from itertools import cycle

async def fetch_url(session, url, proxy, semaphore):
    async with semaphore:  # Begrenzung der gleichzeitigen Anfragen
        try:
            async with session.get(url, proxy=proxy, timeout=aiohttp.ClientTimeout(total=10)) as response:
                data = await response.json()
                return {'url': url, 'ip': data.get('origin'), 'status': response.status}
        except Exception as e:
            return {'url': url, 'error': str(e)}

async def main():
    urls = [f'https://httpbin.org/ip' for _ in range(50)]  # 50 Anfragen
    proxy_list = [
        'http://user:pass@proxy1.example.com:8080',
        'http://user:pass@proxy2.example.com:8080',
    ]
    proxy_cycle = cycle(proxy_list)
    
    # Begrenzung: nicht mehr als 10 gleichzeitige Anfragen
    semaphore = asyncio.Semaphore(10)
    
    async with aiohttp.ClientSession() as session:
        tasks = [
            fetch_url(session, url, next(proxy_cycle), semaphore)
            for url in urls
        ]
        results = await asyncio.gather(*tasks)
        
        # Analyse der Ergebnisse
        successful = [r for r in results if 'ip' in r]
        failed = [r for r in results if 'error' in r]
        
        print(f"Erfolgreiche Anfragen: {len(successful)}")
        print(f"Fehlerhafte Anfragen: {len(failed)}")

asyncio.run(main())

Die Verwendung von asyncio.Semaphore ist entscheidend bei der Arbeit mit Proxys — eine zu hohe Anzahl gleichzeitiger Verbindungen über eine IP kann zu einer Sperrung durch die Zielwebsite oder den Proxy-Anbieter führen.

Fehler- und Timeout-Behandlung

Die Arbeit mit Proxys ist mit einer erhöhten Anzahl von Fehlern verbunden: Timeouts, Verbindungsabbrüche, Ausfälle von Proxy-Servern. Eine korrekte Fehlerbehandlung ist der Schlüssel zur Stabilität des Parsers.

Typische Fehler bei der Arbeit mit Proxys:

Fehler Ursache Lösung
ProxyError Proxy-Server nicht verfügbar Auf einen anderen Proxy umschalten
ConnectTimeout Proxy antwortet nicht rechtzeitig Timeout erhöhen oder Proxy wechseln
ProxyAuthenticationRequired Falscher Benutzername/Passwort Anmeldedaten überprüfen
SSLError Probleme mit dem SSL-Zertifikat SSL-Überprüfung deaktivieren (nicht empfohlen)
TooManyRedirects Proxy erzeugt eine Redirect-Schleife Proxy wechseln oder Redirects begrenzen

Fehlerbehandlung in requests:

import requests
from requests.exceptions import ProxyError, ConnectTimeout, RequestException

def fetch_with_retry(url, proxies, max_retries=3):
    for attempt in range(max_retries):
        try:
            response = requests.get(
                url, 
                proxies=proxies, 
                timeout=(5, 10),  # (connect timeout, read timeout)
                allow_redirects=True,
                verify=True  # SSL-Zertifikat überprüfen
            )
            response.raise_for_status()  # Wirft eine Ausnahme bei 4xx/5xx
            return response.json()
            
        except ProxyError as e:
            print(f"Versuch {attempt + 1}: Proxy nicht verfügbar - {e}")
        except ConnectTimeout as e:
            print(f"Versuch {attempt + 1}: Verbindungs-Timeout - {e}")
        except requests.exceptions.HTTPError as e:
            print(f"Versuch {attempt + 1}: HTTP-Fehler {e.response.status_code}")
            if e.response.status_code == 407:  # Proxy Authentication Required
                print("Proxy-Authentifizierungsfehler!")
                break  # Bei Authentifizierungsfehler nicht wiederholen
        except RequestException as e:
            print(f"Versuch {attempt + 1}: Allgemeiner Fehler - {e}")
        
        if attempt < max_retries - 1:
            print(f"Wiederholen in 2 Sekunden...")
            import time
            time.sleep(2)
    
    raise Exception(f"Anfrage konnte nach {max_retries} Versuchen nicht ausgeführt werden")

# Verwendung
proxies = {'http': 'http://user:pass@proxy.example.com:8080', 'https': 'http://user:pass@proxy.example.com:8080'}
try:
    data = fetch_with_retry('https://httpbin.org/ip', proxies)
    print(data)
except Exception as e:
    print(f"Kritischer Fehler: {e}")

Fehlerbehandlung in aiohttp:

import aiohttp
import asyncio
from aiohttp import ClientError, ClientProxyConnectionError

async def fetch_with_retry(session, url, proxy, max_retries=3):
    for attempt in range(max_retries):
        try:
            timeout = aiohttp.ClientTimeout(total=10, connect=5)
            async with session.get(url, proxy=proxy, timeout=timeout) as response:
                response.raise_for_status()
                return await response.json()
                
        except ClientProxyConnectionError as e:
            print(f"Versuch {attempt + 1}: Fehler beim Verbinden mit Proxy - {e}")
        except asyncio.TimeoutError:
            print(f"Versuch {attempt + 1}: Timeout")
        except aiohttp.ClientHttpProxyError as e:
            print(f"Versuch {attempt + 1}: HTTP-Fehler des Proxys - {e}")
            if e.status == 407:
                print("Proxy-Authentifizierungsfehler!")
                break
        except ClientError as e:
            print(f"Versuch {attempt + 1}: Allgemeiner Client-Fehler - {e}")
        
        if attempt < max_retries - 1:
            await asyncio.sleep(2)
    
    raise Exception(f"Anfrage konnte nach {max_retries} Versuchen nicht ausgeführt werden")

async def main():
    proxy = 'http://user:pass@proxy.example.com:8080'
    async with aiohttp.ClientSession() as session:
        try:
            data = await fetch_with_retry(session, 'https://httpbin.org/ip', proxy)
            print(data)
        except Exception as e:
            print(f"Kritischer Fehler: {e}")

asyncio.run(main())

Timeouts einrichten:

Die richtige Einstellung von Timeouts ist entscheidend für die Stabilität. Empfohlene Werte:

  • Connect-Timeout: 5-10 Sekunden (Zeit für die Verbindung mit dem Proxy)
  • Read-Timeout: 10-30 Sekunden (Zeit für den Erhalt einer Antwort von der Zielwebsite)
  • Gesamt-Timeout: 30-60 Sekunden (Gesamtzeit der Anfrage)

Für langsame residential Proxys wird empfohlen, die Timeouts auf 20-30 Sekunden pro Verbindung zu erhöhen, da die Routenführung über reale Anbieter mehr Zeit in Anspruch nehmen kann.

Best Practices und Optimierung

Eine effektive Arbeit mit Proxys erfordert die Einhaltung einer Reihe von Regeln zur Minimierung von Sperrungen und Maximierung der Leistung.

1. Verwendung von Session zur Wiederverwendung von Verbindungen:

# requests: Session wiederverwendet TCP-Verbindungen
session = requests.Session()
session.proxies = {'http': proxy, 'https': proxy}

for url in urls:
    response = session.get(url)  # Schneller als requests.get()

# aiohttp: Session ist für Asynchronität erforderlich
async with aiohttp.ClientSession() as session:
    tasks = [session.get(url, proxy=proxy) for url in urls]
    await asyncio.gather(*tasks)

2. Realistische User-Agent- und Header-Einstellungen:

import requests

headers = {
    'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36',
    'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8',
    'Accept-Language': 'en-US,en;q=0.5',
    'Accept-Encoding': 'gzip, deflate, br',
    'DNT': '1',
    'Connection': 'keep-alive',
    'Upgrade-Insecure-Requests': '1'
}

proxies = {'http': proxy, 'https': proxy}
response = requests.get('https://example.com', headers=headers, proxies=proxies)

3. Begrenzung der Rate-Limits (Anfragen pro Sekunde):

import time
import requests

class RateLimiter:
    def __init__(self, max_requests_per_second):
        self.max_requests = max_requests_per_second
        self.interval = 1.0 / max_requests_per_second
        self.last_request_time = 0
    
    def wait(self):
        elapsed = time.time() - self.last_request_time
        if elapsed < self.interval:
            time.sleep(self.interval - elapsed)
        self.last_request_time = time.time()

# Verwendung: nicht mehr als 2 Anfragen pro Sekunde
limiter = RateLimiter(2)
proxies = {'http': proxy, 'https': proxy}

for url in urls:
    limiter.wait()
    response = requests.get(url, proxies=proxies)

4. Proxys überwachen und protokollieren:

import logging
from collections import defaultdict

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class ProxyMonitor:
    def __init__(self):
        self.stats = defaultdict(lambda: {'success': 0, 'failed': 0, 'total_time': 0})
    
    def log_request(self, proxy, success, response_time):
        stats = self.stats[proxy]
        if success:
            stats['success'] += 1
        else:
            stats['failed'] += 1
        stats['total_time'] += response_time
        
        # Protokollierung alle 10 Anfragen
        total = stats['success'] + stats['failed']
        if total % 10 == 0:
            avg_time = stats['total_time'] / total
            success_rate = stats['success'] / total * 100
            logger.info(f"Proxy {proxy}: {total} Anfragen, Erfolg {success_rate:.1f}%, avg {avg_time:.2f}s")

monitor = ProxyMonitor()

# Im Anfragecode
import time
start = time.time()
try:
    response = requests.get(url, proxies=proxies, timeout=10)
    monitor.log_request(proxy, True, time.time() - start)
except Exception as e:
    monitor.log_request(proxy, False, time.time() - start)
    logger.error(f"Fehler mit Proxy {proxy}: {e}")

5. DNS-Caching zur Beschleunigung:

# aiohttp mit DNS-Caching
import aiohttp
from aiohttp.resolver import AsyncResolver

resolver = AsyncResolver(nameservers=['8.8.8.8', '8.8.4.4'])
connector = aiohttp.TCPConnector(resolver=resolver, ttl_dns_cache=300)

async with aiohttp.ClientSession(connector=connector) as session:
    # Anfragen verwenden DNS-Cache für 5 Minuten
    async with session.get(url, proxy=proxy) as response:
        data = await response.json()

6. Umgang mit Captchas und Sperrungen:

Tipp: Bei Erhalt des Status 403, 429 oder Captchas wird empfohlen:

  • Proxy auf eine IP aus einem anderen Subnetz wechseln
  • Verzögerung zwischen Anfragen erhöhen (bis zu 5-10 Sekunden)
  • User-Agent und andere Header ändern
  • Cookies aus vorherigen erfolgreichen Sitzungen verwenden

Vergleich von requests und aiohttp für Proxys

Die Wahl zwischen requests und aiohttp hängt von der Aufgabe und dem Datenvolumen ab. Lassen Sie uns die wichtigsten Unterschiede betrachten.

Kriterium requests aiohttp
Synchronität Synchron (blockierend) Asynchron (nicht blockierend)
Leistung ~10-50 Anfragen/Sekunde ~100-1000 Anfragen/Sekunde
Code Einfachheit Einfacher für Anfänger Erfordert Kenntnisse von async/await
Proxy-Einrichtung Dictionary proxies Parameter proxy
SOCKS5 Unterstützung Über requests[socks] Über aiohttp-socks
Speichernutzung Weniger (ein Thread) Mehr (viele Aufgaben)
Besser für Einfache Skripte, <100 Anfragen Parser, >1000 Anfragen

Wann requests verwenden:

  • Einfache Skripte für einmalige Aufgaben
  • Prototyping und Testing
  • Geringes Anfragevolumen (bis zu 100 pro Minute)
  • Wenn Code-Einfachheit und Lesbarkeit wichtig sind
  • Integration mit synchronen Bibliotheken

Wann aiohttp verwenden:

  • Parsing großer Datenmengen (Tausende von Seiten)
  • Überwachung vieler Quellen in Echtzeit
  • API-Dienste mit hoher Last
  • Wenn die Verarbeitungsgeschwindigkeit entscheidend ist
  • Arbeiten mit WebSocket über Proxys

Praktischer Vergleich der Leistung:

# Test: 100 Anfragen über Proxy

# requests (synchron) - ~50 Sekunden
import requests
import time

start = time.time()
proxies = {'http': proxy, 'https': proxy}
for i in range(100):
    response = requests.get('https://httpbin.org/ip', proxies=proxies)
print(f"requests: {time.time() - start:.2f} Sekunden")

# aiohttp (asynchron) - ~5 Sekunden
import aiohttp
import asyncio

async def fetch_all():
    async with aiohttp.ClientSession() as session:
        tasks = [
            session.get('https://httpbin.org/ip', proxy=proxy)
            for _ in range(100)
        ]
        await asyncio.gather(*tasks)

start = time.time()
asyncio.run(fetch_all())
print(f"aiohttp: {time.time() - start:.2f} Sekunden")

Bei der Verwendung von Datacenter-Proxys für schnelles Parsing zeigt aiohttp einen Vorteil von 10-20 Mal im Vergleich zu requests dank der parallelen Verarbeitung von Anfragen.

Fazit

Die Einrichtung von Proxys in Python über die Bibliotheken requests und aiohttp ist eine grundlegende Fähigkeit für die Entwicklung von Parsern, die Automatisierung der Datensammlung und die Umgehung geografischer Einschränkungen. Die Bibliothek requests eignet sich für einfache Skripte und Prototyping aufgrund ihrer verständlichen synchronen API, während aiohttp eine hohe Leistung bei der Verarbeitung von Tausenden von Anfragen über eine asynchrone Architektur bietet.

Wichtige Punkte für eine effektive Arbeit mit Proxys in Python: korrekte Fehler- und Timeout-Behandlung, Implementierung von IP-Rotation zur Lastverteilung, Verwendung von Session zur Wiederverwendung von Verbindungen, Einrichtung realistischer Header und User-Agent, Überwachung der Leistung von Proxy-Servern. Für SOCKS5-Proxys sind zusätzliche Bibliotheken erforderlich — requests[socks] oder aiohttp-socks.

Bei der Auswahl des Proxy-Typs für das Parsing sollten Sie die Spezifik der Aufgabe berücksichtigen: Für stark belastete Parser mit Tausenden von Anfragen eignen sich schnelle Datacenter-Proxys, während für die Umgehung strenger Anti-Bot-Systeme und die Arbeit mit sozialen Netzwerken residential Proxys mit realen IPs von Haushaltsnutzern empfohlen werden. Für Aufgaben, die maximale Anonymität und die Imitation von mobilem Verkehr erfordern, sind mobile Proxys mit IPs von Mobilfunkanbietern optimal.

Wenn Sie planen, leistungsstarke Parser zu entwickeln oder die Datensammlung aus vielen Quellen zu automatisieren, empfehlen wir, residential Proxys auszuprobieren — sie bieten ein hohes Maß an Anonymität, minimales Risiko von Sperrungen und eine stabile Arbeit mit den meisten geschützten Webdiensten. Für technische Aufgaben mit hoher Verarbeitungsgeschwindigkeit eignen sich auch Datacenter-Proxys mit niedriger Latenz und hoher Bandbreite.

```