Intégration de proxy avec Google Cloud Functions : configuration pour le scraping et l'automatisation

Google Cloud Functions est une plateforme sans serveur pour exécuter du code sans gestion de serveurs. Lors du scraping, de l'automatisation des requêtes API ou de la collecte de données, il est souvent nécessaire de router le trafic via un proxy pour contourner les blocages, faire de la rotation d'IP et cibler géographiquement. Dans ce guide, nous allons examiner la configuration d'un proxy dans Cloud Functions en Python et Node.js avec des exemples pratiques.

Pourquoi utiliser un proxy dans Cloud Functions

Google Cloud Functions fonctionne dans un environnement isolé avec des adresses IP partagées des centres de données Google. Lors de requêtes fréquentes vers des API externes ou des sites web, des problèmes surviennent :

• Blocages par IP — de nombreux services (Google, Facebook, marketplaces) reconnaissent le trafic des centres de données et appliquent des limitations de taux ou un blocage total.
• Restrictions géographiques — pour accéder à du contenu disponible uniquement dans certains pays (par exemple, le scraping des prix régionaux sur Wildberries ou Ozon).
• Limitation de taux — une adresse IP peut effectuer un nombre limité de requêtes par minute. Les proxies permettent de répartir la charge.
• Anonimité — masquer la véritable source des requêtes lors de la manipulation de données sensibles ou d'intelligence concurrentielle.

Scénarios typiques d'utilisation de proxies dans Cloud Functions :

• Scraping des marketplaces (Wildberries, Ozon, Amazon) pour surveiller les prix des concurrents
• Collecte de données à partir des réseaux sociaux (Instagram, TikTok) via API ou web scraping
• Automatisation de la vérification des annonces publicitaires dans différentes régions
• Requêtes massives vers des moteurs de recherche (Google, Yandex) pour l'analyse SEO
• Test des fonctionnalités de géolocalisation des applications

Quels types de proxy conviennent à Cloud Functions

Le choix du type de proxy dépend de la tâche, du budget et des exigences en matière d'anonymat. Voici une comparaison des principales options :

Recommandations pour le choix :

• Pour le scraping des marketplaces (Wildberries, Ozon, Amazon) — proxies résidentiels avec rotation à la demande, afin que chaque requête provienne d'une nouvelle IP.
• Pour les requêtes API (Google Maps API, OpenWeatherMap) — proxies de centre de données avec une vitesse élevée, s'il n'y a pas de restrictions strictes sur les IP.
• Pour les réseaux sociaux (Instagram, TikTok) — proxies mobiles, car ils possèdent des IP d'opérateurs mobiles et sont rarement bloqués.
• Pour le scraping SEO (Google, Yandex) — proxies résidentiels avec une attache géographique à la région souhaitée.

Configuration du proxy en Python (requests, aiohttp)

Python est le langage le plus populaire pour Cloud Functions lors du scraping et de l'automatisation. Examinons l'intégration de proxy avec les bibliothèques requests (requêtes synchrones) et aiohttp (requêtes asynchrones).

Exemple avec la bibliothèque requests (proxy HTTP)

import requests
import os

def parse_with_proxy(request):
    # Obtention des données du proxy à partir des variables d'environnement
    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')
    
    # Construction de l'URL du proxy avec authentification
    proxy_url = f"http://{proxy_user}:{proxy_pass}@{proxy_host}:{proxy_port}"
    
    proxies = {
        'http': proxy_url,
        'https': proxy_url
    }
    
    try:
        # Effectuer une requête via le proxy avec un délai d'attente
        response = requests.get(
            'https://api.example.com/data',
            proxies=proxies,
            timeout=10,
            headers={'User-Agent': 'Mozilla/5.0'}
        )
        
        # Vérifier le statut de la réponse
        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'Erreur de proxy : {str(e)}'}
    except requests.exceptions.Timeout:
        return {'statusCode': 504, 'error': 'Délai d\'attente de la requête'}
    except requests.exceptions.RequestException as e:
        return {'statusCode': 500, 'error': f'Échec de la requête : {str(e)}'}

Points importants :

• Variables d'environnement — stockez les données du proxy (hôte, port, login, mot de passe) dans le Secret Manager ou les variables d'environnement de Cloud Functions, et non dans le code.
• Délai d'attente — assurez-vous de définir un timeout, afin que la fonction ne se bloque pas en cas de problème avec le proxy.
• User-Agent — ajoutez l'en-tête User-Agent pour que les requêtes ressemblent à celles d'un véritable navigateur.
• Gestion des erreurs — traitez séparément ProxyError (problèmes de proxy) et Timeout (proxy lent).

Exemple avec aiohttp (requêtes asynchrones)

Pour les tâches à forte charge (par exemple, le scraping de 1000+ pages), utilisez des requêtes asynchrones avec 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': 'Échec de la connexion au proxy'}
        except asyncio.TimeoutError:
            return {'error': 'Délai d\'attente de la requête'}

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'
    ]
    
    # Lancer des requêtes asynchrones en parallèle
    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}

L'approche asynchrone permet d'effectuer 10 à 100 requêtes parallèles via le proxy, ce qui est crucial pour le scraping de grands volumes de données dans le cadre d'une durée d'exécution limitée des Cloud Functions (jusqu'à 9 minutes).

Travail avec des proxies SOCKS5

Certains fournisseurs de proxy proposent des proxies SOCKS5 pour un fonctionnement plus fiable avec le trafic UDP ou pour contourner les blocages. Pour travailler avec SOCKS5 en Python, utilisez la bibliothèque requests[socks] :

# Ajoutez dans requirements.txt :
# 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()}

Configuration du proxy en Node.js (axios, node-fetch)

Node.js est le deuxième langage le plus populaire pour Cloud Functions. Examinons l'intégration de proxy avec les bibliothèques axios et node-fetch.

Exemple avec 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] // Retourne l'hôte:port sans mot de passe
    });
    
  } catch (error) {
    if (error.code === 'ECONNREFUSED') {
      res.status(502).json({ error: 'Connexion au proxy refusée' });
    } else if (error.code === 'ETIMEDOUT') {
      res.status(504).json({ error: 'Délai d\'attente du proxy' });
    } else {
      res.status(500).json({ error: error.message });
    }
  }
};

Dépendances pour package.json :

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

Exemple avec node-fetch et 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 });
  }
};

Dépendances pour SOCKS5 :

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

Authentification du proxy : login/mot de passe et liste blanche d'IP

Il existe deux méthodes principales d'authentification lors de l'utilisation de proxies :

1. Authentification par login et mot de passe

La méthode la plus courante consiste à transmettre les informations d'identification dans l'URL du proxy :

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

Avantages : Simplicité de configuration, ne nécessite pas d'IP source fixe.

Inconvénients : Les informations d'identification sont transmises à chaque requête, léger overhead.

2. Authentification par liste blanche d'IP

Certains fournisseurs permettent d'ajouter les adresses IP de Cloud Functions à la liste blanche. Problème : Cloud Functions utilise des IP dynamiques du pool Google Cloud.

Solution : Utilisez Cloud NAT pour router le trafic sortant via une IP externe statique :

1. Créez un réseau VPC et un sous-réseau dans Google Cloud
2. Configurez Cloud NAT avec réservation d'une IP statique
3. Connectez Cloud Functions au VPC Connector
4. Ajoutez l'IP statique à la liste blanche du fournisseur de proxy

Après la configuration, le proxy ne nécessite pas de login et de mot de passe :

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

Recommandation : Pour la plupart des cas, utilisez l'authentification par login/mot de passe — c'est plus simple et ne nécessite pas de coûts supplémentaires pour Cloud NAT (à partir de 0,044 $/heure + trafic).

Rotation des IP et gestion du pool de proxy

Lors du scraping de grands volumes de données, il est crucial d'utiliser la rotation des IP pour éviter les blocages. Plusieurs approches existent :

1. Rotation côté fournisseur (Rotating Proxies)

De nombreux fournisseurs proposent des proxies rotatifs — un seul point de terminaison qui change automatiquement d'IP à chaque requête ou par minute :

# Un seul point de terminaison, l'IP change automatiquement
proxy_url = "http://username:password@rotating.proxy.com:8080"

# Chaque requête provient d'une nouvelle IP
for i in range(100):
    response = requests.get('https://api.ipify.org', proxies={'http': proxy_url})
    print(f"Requête {i}: IP = {response.text}")

Avantages : Pas besoin de gérer manuellement le pool de proxies, intégration simple.

Inconvénients : Pas de contrôle sur des IP spécifiques, peut être plus coûteux.

2. Gestion manuelle du pool de proxy

Si vous avez une liste de proxies statiques, implémentez la rotation au niveau du code :

import random
import requests

# Pool de proxies (peut être chargé depuis le Secret Manager)
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:
            # Si le proxy ne fonctionne pas, essayez un autre
            proxy = get_random_proxy()
            response = requests.get(url, proxies={'http': proxy, 'https': proxy})
            results.append({'url': url, 'status': response.status_code})
    
    return results

3. Proxies basés sur les sessions (sessions collantes)

Pour les tâches où il est nécessaire de conserver une IP au sein d'une session (par exemple, l'authentification sur un site), utilisez l'ID de session dans l'URL du proxy :

# Ajout de l'ID de session dans le login
import uuid

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

# Toutes les requêtes avec cet ID de session passeront par une seule IP
session = requests.Session()
session.proxies = {'http': proxy_url, 'https': proxy_url}

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

# Requêtes suivantes dans la même session
session.get('https://example.com/dashboard')

Gestion des erreurs et délais d'attente

Lors de l'utilisation de proxies dans Cloud Functions, il est crucial de gérer correctement les erreurs pour ne pas perdre de données et ne pas dépasser les limites de temps d'exécution.

Types d'erreurs et méthodes de gestion

Exemple de gestion complète des erreurs

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

def fetch_with_retry(url, proxy_pool, max_retries=3):
    """
    Requête avec retry automatique et changement de proxy en cas d'erreurs
    """
    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'}
            )
            
            # Vérifier le code de statut
            if response.status_code == 200:
                return {'success': True, 'data': response.text, 'proxy': proxy}
            elif response.status_code == 429:
                # Limitation de taux — attendre et essayer à nouveau
                time.sleep(2 ** attempt)  # Backoff exponentiel
                continue
            elif response.status_code == 403:
                # IP bloquée — changer de proxy
                continue
            else:
                return {'success': False, 'status': response.status_code}
                
        except ProxyError:
            # Proxy ne fonctionne pas — essayer suivant
            print(f"Proxy {proxy} échoué, essayant un autre...")
            continue
        except Timeout:
            # Délai d'attente — essayer avec un autre proxy
            print(f"Délai d'attente avec {proxy}, réessayer...")
            continue
        except RequestException as e:
            # Autres erreurs
            print(f"Échec de la requête : {e}")
            if attempt == max_retries - 1:
                return {'success': False, 'error': str(e)}
            continue
    
    return {'success': False, 'error': 'Nombre maximum de tentatives dépassé'}

Configuration des délais d'attente dans Cloud Functions

Les Cloud Functions ont une limite de temps d'exécution (par défaut 60 secondes, maximum 540 secondes). Prenez cela en compte lors de la configuration des délais d'attente des proxies :

• Délai de connexion — temps pour établir une connexion avec le proxy (recommandé 5 secondes)
• Délai de lecture — temps pour obtenir une réponse du serveur cible via le proxy (recommandé 10-15 secondes)
• Délai total — temps total pour toute la requête (doit être inférieur au délai d'attente de la fonction)

# Python : délais d'attente séparés
response = requests.get(
    url,
    proxies=proxies,
    timeout=(5, 15)  # (délai de connexion, délai de lecture)
)

# Node.js avec axios
const response = await axios.get(url, {
  httpsAgent: agent,
  timeout: 10000  // délai total en millisecondes
});

Meilleures pratiques et optimisation des performances

Recommandations pour un fonctionnement efficace avec des proxies dans Cloud Functions :

1. Utilisez des variables d'environnement pour les informations d'identification

Ne stockez jamais les logins et mots de passe des proxies dans le code. Utilisez le Secret Manager ou des variables d'environnement :

# Création d'un secret dans Google Cloud
gcloud secrets create proxy-credentials \
    --data-file=proxy-config.json

# Fournir l'accès aux Cloud Functions
gcloud secrets add-iam-policy-binding proxy-credentials \
    --member=serviceAccount:PROJECT_ID@appspot.gserviceaccount.com \
    --role=roles/secretmanager.secretAccessor

# Lecture du secret dans le 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. Mettez en cache les résultats du scraping

Utilisez Cloud Storage ou Firestore pour mettre en cache les données, afin de ne pas effectuer de requêtes répétées via le proxy :

import hashlib
from google.cloud import storage

def fetch_with_cache(url, proxy):
    # Générer une clé de cache basée sur l'URL
    cache_key = hashlib.md5(url.encode()).hexdigest()
    
    # Vérifier le cache dans Cloud Storage
    bucket = storage.Client().bucket('my-cache-bucket')
    blob = bucket.blob(f"cache/{cache_key}.json")
    
    if blob.exists():
        # Retourner les données mises en cache
        return json.loads(blob.download_as_text())
    
    # Effectuer une requête via le proxy
    response = requests.get(url, proxies={'http': proxy})
    data = response.json()
    
    # Sauvegarder dans le cache
    blob.upload_from_string(json.dumps(data))
    
    return data

3. Surveillance et journalisation

Suivez les performances des proxies et la fréquence des erreurs via 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. Optimisation du démarrage à froid

Les Cloud Functions ont un délai de démarrage à froid (cold start). Minimisez les dépendances et utilisez les versions minimales des bibliothèques :

# requirements.txt — uniquement les bibliothèques nécessaires
requests==2.31.0
# Évitez les bibliothèques lourdes comme pandas, si elles ne sont pas critiques

Utilisez des variables globales pour réutiliser les connexions :

# Créez une session une fois lors du démarrage à froid
session = requests.Session()
session.proxies = {'http': PROXY_URL, 'https': PROXY_URL}

def parse_data(request):
    # Réutiliser la session entre les appels
    response = session.get('https://api.example.com/data')
    return response.json()

5. Ciblage géographique des proxies

Pour les tâches avec ciblage géographique (par exemple, scraping des prix régionaux), utilisez des proxies liés à un pays ou une ville spécifiques :

# Exemple avec des proxies résidentiels, où l'on peut spécifier le pays dans le login
proxy_url = f"http://username-country-ru:password@proxy.example.com:8080"

# Ou utiliser différents points de terminaison pour différents pays
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

Conclusion

L'intégration de proxies avec Google Cloud Functions ouvre de larges possibilités pour le scraping, l'automatisation et le travail avec des API sans restrictions d'IP. Les points principaux à considérer : gestion correcte des erreurs avec une logique de retry, utilisation de délais d'attente pour éviter les blocages, rotation des IP pour éviter les blocages et stockage sécurisé des informations d'identification dans le Secret Manager.

Pour la plupart des tâches de scraping et d'automatisation, le choix optimal sera les proxies résidentiels — ils offrent une grande anonymat et un faible taux de blocage grâce à l'utilisation d'IP de véritables utilisateurs. Pour travailler avec des réseaux sociaux et des applications mobiles, nous recommandons les proxies mobiles, qui possèdent des IP d'opérateurs mobiles et sont pratiquement jamais bloqués par des plateformes comme Instagram et TikTok.

Avec une configuration correcte des Cloud Functions avec des proxies, vous obtenez une solution évolutive et économique pour traiter de grands volumes de données sans avoir à gérer l'infrastructure.