Integration von Proxys mit Google Cloud Functions: Einrichtung für Parsing und Automatisierung

Google Cloud Functions ist eine serverlose Plattform zum Ausführen von Code ohne Serververwaltung. Bei der Arbeit mit Parsing, der Automatisierung von API-Anfragen oder der Datensammlung ist es oft erforderlich, den Datenverkehr über Proxys zu leiten, um Sperren zu umgehen, IPs zu rotieren und geografisches Targeting zu ermöglichen. In diesem Leitfaden werden wir die Einrichtung von Proxys in Cloud Functions mit Python und Node.js anhand praktischer Beispiele erläutern.

Warum Proxys in Cloud Functions verwenden

Google Cloud Functions arbeiten in einer isolierten Umgebung mit gemeinsamen IP-Adressen der Google-Rechenzentren. Bei häufigen Anfragen an externe APIs oder Websites treten folgende Probleme auf:

• IP-Sperren – viele Dienste (Google, Facebook, Marktplätze) erkennen den Datenverkehr aus Rechenzentren und wenden Rate-Limiting oder vollständige Sperren an.
• Geografische Einschränkungen – um auf Inhalte zuzugreifen, die nur in bestimmten Ländern verfügbar sind (z. B. Parsing regionaler Preise auf Wildberries oder Ozon).
• Rate-Limiting – eine IP-Adresse kann eine begrenzte Anzahl von Anfragen pro Minute stellen. Proxys ermöglichen eine Lastverteilung.
• Anonymität – Verbergen der tatsächlichen Quelle von Anfragen bei der Arbeit mit sensiblen Daten oder Wettbewerbsanalysen.

Typische Anwendungsfälle für Proxys in Cloud Functions:

• Parsing von Marktplätzen (Wildberries, Ozon, Amazon) zur Überwachung von Wettbewerberpreisen
• Datensammlung aus sozialen Netzwerken (Instagram, TikTok) über APIs oder Web-Scraping
• Automatisierung der Überprüfung von Werbeanzeigen in verschiedenen Regionen
• Massenanfragen an Suchmaschinen (Google, Yandex) für SEO-Analysen
• Testen von Geolokalisierungsfunktionen von Anwendungen

Welche Proxy-Typen für Cloud Functions geeignet sind

Die Wahl des Proxy-Typs hängt von der Aufgabe, dem Budget und den Anforderungen an die Anonymität ab. Hier ist ein Vergleich der wichtigsten Optionen:

Empfehlungen zur Auswahl:

• Für das Parsing von Marktplätzen (Wildberries, Ozon, Amazon) – residential Proxys mit Rotation pro Anfrage, damit jede Anfrage von einer neuen IP kommt.
• Für API-Anfragen (Google Maps API, OpenWeatherMap) – datacenter Proxys mit hoher Geschwindigkeit, wenn es keine strengen IP-Beschränkungen gibt.
• Für soziale Netzwerke (Instagram, TikTok) – mobile Proxys, da sie IPs von Mobilfunkanbietern haben und selten blockiert werden.
• Für SEO-Parsings (Google, Yandex) – residential Proxys mit geografischer Bindung an die benötigte Region.

Proxy in Python einrichten (requests, aiohttp)

Python ist die beliebteste Sprache für Cloud Functions bei der Arbeit mit Parsing und Automatisierung. Lassen Sie uns die Integration von Proxys mit den Bibliotheken requests (synchronisierte Anfragen) und aiohttp (asynchrone Anfragen) betrachten.

Beispiel mit der Bibliothek requests (HTTP-Proxy)

import requests
import os

def parse_with_proxy(request):
    # Proxy-Daten aus Umgebungsvariablen abrufen
    proxy_host = os.environ.get('PROXY_HOST', 'proxy.example.com')
    proxy_port = os.environ.get('PROXY_PORT', '8080')
    proxy_user = os.environ.get('PROXY_USER', 'username')
    proxy_pass = os.environ.get('PROXY_PASS', 'password')
    
    # Proxy-URL mit Authentifizierung erstellen
    proxy_url = f"http://{proxy_user}:{proxy_pass}@{proxy_host}:{proxy_port}"
    
    proxies = {
        'http': proxy_url,
        'https': proxy_url
    }
    
    try:
        # Anfrage über Proxy mit Timeout stellen
        response = requests.get(
            'https://api.example.com/data',
            proxies=proxies,
            timeout=10,
            headers={'User-Agent': 'Mozilla/5.0'}
        )
        
        # Status des Antwort überprüfen
        response.raise_for_status()
        
        return {
            'statusCode': 200,
            'body': response.json(),
            'ip_used': response.headers.get('X-Forwarded-For', 'unknown')
        }
        
    except requests.exceptions.ProxyError as e:
        return {'statusCode': 502, 'error': f'Proxy-Fehler: {str(e)}'}
    except requests.exceptions.Timeout:
        return {'statusCode': 504, 'error': 'Anfrage-Timeout'}
    except requests.exceptions.RequestException as e:
        return {'statusCode': 500, 'error': f'Anfrage fehlgeschlagen: {str(e)}'}

Wichtige Punkte:

• Umgebungsvariablen – speichern Sie die Proxy-Daten (Host, Port, Benutzername, Passwort) im Secret Manager oder in den Umgebungsvariablen von Cloud Functions, nicht im Code.
• Timeouts – setzen Sie unbedingt timeout, damit die Funktion nicht hängt, wenn es Probleme mit dem Proxy gibt.
• User-Agent – fügen Sie den User-Agent-Header hinzu, damit die Anfragen wie von einem echten Browser aussehen.
• Fehlerbehandlung – behandeln Sie ProxyError (Probleme mit dem Proxy) und Timeout (langsamer Proxy) separat.

Beispiel mit aiohttp (asynchrone Anfragen)

Für hochbelastete Aufgaben (z. B. Parsing von 1000+ Seiten) verwenden Sie asynchrone Anfragen mit aiohttp:

import aiohttp
import asyncio
import os

async def fetch_with_proxy(url, proxy_url):
    async with aiohttp.ClientSession() as session:
        try:
            async with session.get(
                url,
                proxy=proxy_url,
                timeout=aiohttp.ClientTimeout(total=10),
                headers={'User-Agent': 'Mozilla/5.0'}
            ) as response:
                return await response.text()
        except aiohttp.ClientProxyConnectionError:
            return {'error': 'Proxy-Verbindung fehlgeschlagen'}
        except asyncio.TimeoutError:
            return {'error': 'Anfrage-Timeout'}

def parse_multiple_urls(request):
    proxy_url = f"http://{os.environ['PROXY_USER']}:{os.environ['PROXY_PASS']}@{os.environ['PROXY_HOST']}:{os.environ['PROXY_PORT']}"
    
    urls = [
        'https://example.com/page1',
        'https://example.com/page2',
        'https://example.com/page3'
    ]
    
    # Asynchrone Anfragen parallel ausführen
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    
    tasks = [fetch_with_proxy(url, proxy_url) for url in urls]
    results = loop.run_until_complete(asyncio.gather(*tasks))
    
    return {'statusCode': 200, 'results': results}

Der asynchrone Ansatz ermöglicht es, 10-100 parallele Anfragen über Proxys zu stellen, was entscheidend für das Parsen großer Datenmengen innerhalb der begrenzten Ausführungszeit von Cloud Functions (bis zu 9 Minuten) ist.

Arbeiten mit SOCKS5-Proxys

Einige Proxy-Anbieter bieten SOCKS5-Proxys für eine zuverlässigere Arbeit mit UDP-Datenverkehr oder zum Umgehen von Sperren an. Verwenden Sie für SOCKS5 in Python die Bibliothek requests[socks]:

# Fügen Sie in requirements.txt hinzu:
# requests[socks]

import requests

def use_socks5_proxy(request):
    proxy_url = f"socks5://{os.environ['PROXY_USER']}:{os.environ['PROXY_PASS']}@{os.environ['PROXY_HOST']}:{os.environ['PROXY_PORT']}"
    
    proxies = {
        'http': proxy_url,
        'https': proxy_url
    }
    
    response = requests.get(
        'https://api.ipify.org?format=json',
        proxies=proxies,
        timeout=10
    )
    
    return {'statusCode': 200, 'ip': response.json()}

Proxy in Node.js einrichten (axios, node-fetch)

Node.js ist die zweitbeliebteste Sprache für Cloud Functions. Lassen Sie uns die Integration von Proxys mit den Bibliotheken axios und node-fetch betrachten.

Beispiel mit axios

const axios = require('axios');
const HttpsProxyAgent = require('https-proxy-agent');

exports.parseWithProxy = async (req, res) => {
  const proxyUrl = `http://${process.env.PROXY_USER}:${process.env.PROXY_PASS}@${process.env.PROXY_HOST}:${process.env.PROXY_PORT}`;
  
  const agent = new HttpsProxyAgent(proxyUrl);
  
  try {
    const response = await axios.get('https://api.example.com/data', {
      httpsAgent: agent,
      timeout: 10000,
      headers: {
        'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
      }
    });
    
    res.status(200).json({
      success: true,
      data: response.data,
      proxyUsed: proxyUrl.split('@')[1] // Gibt Host:Port ohne Passwort zurück
    });
    
  } catch (error) {
    if (error.code === 'ECONNREFUSED') {
      res.status(502).json({ error: 'Proxy-Verbindung abgelehnt' });
    } else if (error.code === 'ETIMEDOUT') {
      res.status(504).json({ error: 'Proxy-Timeout' });
    } else {
      res.status(500).json({ error: error.message });
    }
  }
};

Abhängigkeiten für package.json:

{
  "dependencies": {
    "axios": "^1.6.0",
    "https-proxy-agent": "^7.0.2"
  }
}

Beispiel mit node-fetch und SOCKS5

const fetch = require('node-fetch');
const { SocksProxyAgent } = require('socks-proxy-agent');

exports.fetchWithSocks5 = async (req, res) => {
  const proxyUrl = `socks5://${process.env.PROXY_USER}:${process.env.PROXY_PASS}@${process.env.PROXY_HOST}:${process.env.PROXY_PORT}`;
  
  const agent = new SocksProxyAgent(proxyUrl);
  
  try {
    const response = await fetch('https://api.ipify.org?format=json', {
      agent,
      timeout: 10000
    });
    
    const data = await response.json();
    
    res.status(200).json({
      success: true,
      yourIP: data.ip
    });
    
  } catch (error) {
    res.status(500).json({ error: error.message });
  }
};

Abhängigkeiten für SOCKS5:

{
  "dependencies": {
    "node-fetch": "^2.7.0",
    "socks-proxy-agent": "^8.0.2"
  }
}

Proxy-Authentifizierung: Benutzername/Passwort und IP-Whitelist

Es gibt zwei Hauptmethoden zur Authentifizierung bei der Arbeit mit Proxys:

1. Authentifizierung mit Benutzername und Passwort

Die am häufigsten verwendete Methode ist die Übertragung von Anmeldedaten in der Proxy-URL:

http://username:password@proxy.example.com:8080

Vorteile: Einfache Einrichtung, erfordert keine feste IP-Quelle.

Nachteile: Anmeldedaten werden bei jeder Anfrage übertragen, geringer Overhead.

2. Authentifizierung über IP-Whitelist

Einige Anbieter erlauben es, die IP-Adressen von Cloud Functions in die Whitelist aufzunehmen. Problem: Cloud Functions verwenden dynamische IPs aus dem Pool von Google Cloud.

Lösung: Verwenden Sie Cloud NAT, um den ausgehenden Datenverkehr über eine statische externe IP zu leiten:

1. Erstellen Sie ein VPC-Netzwerk und ein Subnetz in Google Cloud
2. Richten Sie Cloud NAT mit Reservierung einer statischen IP ein
3. Verbinden Sie Cloud Functions mit dem VPC-Connector
4. Fügen Sie die statische IP zur Whitelist des Proxy-Anbieters hinzu

Nach der Einrichtung benötigt der Proxy keine Anmeldedaten mehr:

proxies = {
    'http': 'http://proxy.example.com:8080',
    'https': 'http://proxy.example.com:8080'
}

Empfehlung: Verwenden Sie in den meisten Fällen die Authentifizierung mit Benutzername/Passwort – das ist einfacher und erfordert keine zusätzlichen Kosten für Cloud NAT (ab $0.044/Stunde + Datenverkehr).

IP-Rotation und Verwaltung des Proxy-Pools

Beim Parsen großer Datenmengen ist es entscheidend, IP-Rotation zu verwenden, um Sperren zu vermeiden. Es gibt mehrere Ansätze:

1. Rotation auf Anbieterseite (Rotating Proxies)

Viele Anbieter bieten rotating Proxys an – einen einzigen Endpunkt, der die IP bei jeder Anfrage oder nach einem Timer automatisch ändert:

# Ein Endpunkt, IP ändert sich automatisch
proxy_url = "http://username:password@rotating.proxy.com:8080"

# Jede Anfrage erfolgt von einer neuen IP
for i in range(100):
    response = requests.get('https://api.ipify.org', proxies={'http': proxy_url})
    print(f"Anfrage {i}: IP = {response.text}")

Vorteile: Kein manuelles Management des Proxy-Pools erforderlich, einfache Integration.

Nachteile: Kein Einfluss auf spezifische IPs, kann teurer sein.

2. Manuelle Verwaltung des Proxy-Pools

Wenn Sie eine Liste statischer Proxys haben, implementieren Sie die Rotation auf Code-Ebene:

import random
import requests

# Proxy-Pool (kann aus Secret Manager geladen werden)
PROXY_POOL = [
    "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():
    return random.choice(PROXY_POOL)

def parse_with_rotation(urls):
    results = []
    
    for url in urls:
        proxy = get_random_proxy()
        
        try:
            response = requests.get(
                url,
                proxies={'http': proxy, 'https': proxy},
                timeout=10
            )
            results.append({
                'url': url,
                'status': response.status_code,
                'proxy': proxy.split('@')[1]
            })
        except Exception as e:
            # Wenn der Proxy nicht funktioniert, versuchen wir einen anderen
            proxy = get_random_proxy()
            response = requests.get(url, proxies={'http': proxy, 'https': proxy})
            results.append({'url': url, 'status': response.status_code})
    
    return results

3. Session-basierte Proxys (sticky sessions)

Für Aufgaben, bei denen eine IP innerhalb einer Sitzung beibehalten werden muss (z. B. Authentifizierung auf einer Website), verwenden Sie die Session-ID in der Proxy-URL:

# Fügen Sie die Session-ID in den Login ein
import uuid

session_id = str(uuid.uuid4())
proxy_url = f"http://username-session-{session_id}:password@proxy.example.com:8080"

# Alle Anfragen mit dieser session_id gehen über eine IP
session = requests.Session()
session.proxies = {'http': proxy_url, 'https': proxy_url}

# Authentifizierung
session.post('https://example.com/login', data={'user': 'test', 'pass': '123'})

# Folgende Anfragen in derselben Sitzung
session.get('https://example.com/dashboard')

Fehlerbehandlung und Timeouts

Bei der Arbeit mit Proxys in Cloud Functions ist es entscheidend, Fehler richtig zu behandeln, um Datenverluste zu vermeiden und die Ausführungszeitlimits nicht zu überschreiten.

Fehlerarten und Behandlungsmethoden

Beispiel für umfassende Fehlerbehandlung

import requests
import time
from requests.exceptions import ProxyError, Timeout, RequestException

def fetch_with_retry(url, proxy_pool, max_retries=3):
    """
    Anfrage mit automatischem Retry und Proxywechsel bei Fehlern
    """
    for attempt in range(max_retries):
        proxy = random.choice(proxy_pool)
        
        try:
            response = requests.get(
                url,
                proxies={'http': proxy, 'https': proxy},
                timeout=10,
                headers={'User-Agent': 'Mozilla/5.0'}
            )
            
            # Statuscode überprüfen
            if response.status_code == 200:
                return {'success': True, 'data': response.text, 'proxy': proxy}
            elif response.status_code == 429:
                # Rate-Limiting – warten und erneut versuchen
                time.sleep(2 ** attempt)  # Exponentielles Backoff
                continue
            elif response.status_code == 403:
                # IP blockiert – Proxy wechseln
                continue
            else:
                return {'success': False, 'status': response.status_code}
                
        except ProxyError:
            # Proxy funktioniert nicht – nächsten versuchen
            print(f"Proxy {proxy} fehlgeschlagen, versuche einen anderen...")
            continue
        except Timeout:
            # Timeout – mit einem anderen Proxy versuchen
            print(f"Timeout mit {proxy}, erneut versuchen...")
            continue
        except RequestException as e:
            # Andere Fehler
            print(f"Anfrage fehlgeschlagen: {e}")
            if attempt == max_retries - 1:
                return {'success': False, 'error': str(e)}
            continue
    
    return {'success': False, 'error': 'Maximale Anzahl an Versuchen überschritten'}

Timeouts in Cloud Functions einstellen

Cloud Functions haben eine Ausführungszeitbeschränkung (standardmäßig 60 Sekunden, maximal 540 Sekunden). Berücksichtigen Sie dies bei der Einstellung der Proxy-Timeouts:

• Connection timeout – Zeit für die Herstellung einer Verbindung zum Proxy (empfohlen 5 Sekunden)
• Read timeout – Zeit für den Erhalt einer Antwort vom Zielserver über den Proxy (empfohlen 10-15 Sekunden)
• Total timeout – Gesamtzeit für die gesamte Anfrage (sollte kürzer sein als das Timeout der Funktion)

# Python: getrennte Timeouts
response = requests.get(
    url,
    proxies=proxies,
    timeout=(5, 15)  # (connect timeout, read timeout)
)

# Node.js mit axios
const response = await axios.get(url, {
  httpsAgent: agent,
  timeout: 10000  // total timeout in Millisekunden
});

Best Practices und Leistungsoptimierung

Empfehlungen für eine effektive Arbeit mit Proxys in Cloud Functions:

1. Verwenden Sie Umgebungsvariablen für Anmeldedaten

Speichern Sie niemals Benutzernamen und Passwörter von Proxys im Code. Verwenden Sie den Secret Manager oder Umgebungsvariablen:

# Erstellen eines Geheimnisses in Google Cloud
gcloud secrets create proxy-credentials \
    --data-file=proxy-config.json

# Zugriff auf Cloud Functions gewähren
gcloud secrets add-iam-policy-binding proxy-credentials \
    --member=serviceAccount:PROJECT_ID@appspot.gserviceaccount.com \
    --role=roles/secretmanager.secretAccessor

# Lesen des Geheimnisses im Code
from google.cloud import secretmanager
import json

def get_proxy_config():
    client = secretmanager.SecretManagerServiceClient()
    name = f"projects/{PROJECT_ID}/secrets/proxy-credentials/versions/latest"
    response = client.access_secret_version(request={"name": name})
    return json.loads(response.payload.data.decode('UTF-8'))

2. Cachen Sie Parsing-Ergebnisse

Verwenden Sie Cloud Storage oder Firestore zum Cachen von Daten, um keine wiederholten Anfragen über Proxys zu stellen:

import hashlib
from google.cloud import storage

def fetch_with_cache(url, proxy):
    # Generieren Sie einen Cache-Schlüssel basierend auf der URL
    cache_key = hashlib.md5(url.encode()).hexdigest()
    
    # Überprüfen Sie den Cache in Cloud Storage
    bucket = storage.Client().bucket('my-cache-bucket')
    blob = bucket.blob(f"cache/{cache_key}.json")
    
    if blob.exists():
        # Geben Sie die zwischengespeicherten Daten zurück
        return json.loads(blob.download_as_text())
    
    # Anfrage über Proxy stellen
    response = requests.get(url, proxies={'http': proxy})
    data = response.json()
    
    # Im Cache speichern
    blob.upload_from_string(json.dumps(data))
    
    return data

3. Überwachung und Protokollierung

Überwachen Sie die Leistung der Proxys und die Fehlerhäufigkeit über Cloud Logging:

import logging
import time

def fetch_with_logging(url, proxy):
    start_time = time.time()
    
    try:
        response = requests.get(url, proxies={'http': proxy}, timeout=10)
        duration = time.time() - start_time
        
        logging.info({
            'url': url,
            'proxy': proxy.split('@')[1],
            'status': response.status_code,
            'duration': duration,
            'success': True
        })
        
        return response
        
    except Exception as e:
        duration = time.time() - start_time
        
        logging.error({
            'url': url,
            'proxy': proxy.split('@')[1],
            'error': str(e),
            'duration': duration,
            'success': False
        })
        
        raise

4. Optimierung des kalten Starts

Cloud Functions haben eine Verzögerung beim kalten Start (cold start). Minimieren Sie Abhängigkeiten und verwenden Sie minimale Versionen von Bibliotheken:

# requirements.txt – nur notwendige Bibliotheken
requests==2.31.0
# Vermeiden Sie schwere Bibliotheken wie pandas, wenn sie nicht kritisch sind

Verwenden Sie globale Variablen zur Wiederverwendung von Verbindungen:

# Erstellen Sie die Sitzung einmal beim kalten Start
session = requests.Session()
session.proxies = {'http': PROXY_URL, 'https': PROXY_URL}

def parse_data(request):
    # Wiederverwendung der Sitzung zwischen Aufrufen
    response = session.get('https://api.example.com/data')
    return response.json()

5. Geografische Bindung von Proxys

Für Aufgaben mit geografischem Targeting (z. B. Parsing regionaler Preise) verwenden Sie Proxys mit Bindung an ein bestimmtes Land oder eine Stadt:

# Beispiel mit residential Proxys, bei denen das Land im Login angegeben werden kann
proxy_url = f"http://username-country-ru:password@proxy.example.com:8080"

# Oder verwenden Sie verschiedene Endpunkte für verschiedene Länder
PROXIES_BY_COUNTRY = {
    'RU': 'http://user:pass@ru.proxy.example.com:8080',
    'US': 'http://user:pass@us.proxy.example.com:8080',
    'DE': 'http://user:pass@de.proxy.example.com:8080'
}

def parse_by_country(country_code):
    proxy = PROXIES_BY_COUNTRY.get(country_code)
    response = requests.get('https://example.com', proxies={'http': proxy})
    return response.text

Fazit

Die Integration von Proxys mit Google Cloud Functions eröffnet umfangreiche Möglichkeiten für Parsing, Automatisierung und API-Arbeit ohne IP-Beschränkungen. Wichtige Punkte, die zu beachten sind: richtige Fehlerbehandlung mit Retry-Logik, Verwendung von Timeouts zur Vermeidung von Hängern, IP-Rotation zur Vermeidung von Sperren und sichere Speicherung von Anmeldedaten im Secret Manager.

Für die meisten Parsing- und Automatisierungsaufgaben sind residential Proxys die optimale Wahl – sie bieten hohe Anonymität und eine niedrige Blockierungsrate durch die Verwendung von IPs realer Benutzer. Für die Arbeit mit sozialen Netzwerken und mobilen Anwendungen empfehlen wir mobile Proxys, die IPs von Mobilfunkanbietern haben und praktisch nicht von Plattformen wie Instagram und TikTok blockiert werden.

Bei richtiger Konfiguration von Cloud Functions mit Proxys erhalten Sie eine skalierbare und kostengünstige Lösung zur Verarbeitung großer Datenmengen, ohne die Infrastruktur verwalten zu müssen.