Integración de proxies con Google Cloud Functions: configuración para scraping y automatización

Google Cloud Functions es una plataforma sin servidor para ejecutar código sin gestionar servidores. Al trabajar con scraping, automatización de solicitudes API o recopilación de datos, a menudo se requiere enrutar el tráfico a través de proxies para eludir bloqueos, rotación de IP y geotargeting. En esta guía, analizaremos la configuración de proxies en Cloud Functions usando Python y Node.js con ejemplos prácticos.

Por qué usar proxies en Cloud Functions

Google Cloud Functions opera en un entorno aislado con direcciones IP compartidas de los centros de datos de Google. Al hacer solicitudes frecuentes a APIs externas o sitios web, surgen problemas:

• Bloqueos por IP — muchos servicios (Google, Facebook, marketplaces) reconocen el tráfico de los centros de datos y aplican limitaciones de tasa o bloqueos completos.
• Restricciones geográficas — para acceder a contenido disponible solo en ciertos países (por ejemplo, scraping de precios regionales en Wildberries o Ozon).
• Limitación de tasa — una dirección IP puede hacer un número limitado de solicitudes por minuto. Los proxies permiten distribuir la carga.
• Anonimato — ocultar la fuente real de las solicitudes al trabajar con datos sensibles o inteligencia competitiva.

Escenarios típicos de uso de proxies en Cloud Functions:

• Scraping de marketplaces (Wildberries, Ozon, Amazon) para monitorear precios de competidores
• Recopilación de datos de redes sociales (Instagram, TikTok) a través de API o web scraping
• Automatización de la verificación de anuncios publicitarios en diferentes regiones
• Solicitudes masivas a motores de búsqueda (Google, Yandex) para análisis SEO
• Pruebas de funciones de geolocalización de aplicaciones

Qué tipos de proxies son adecuados para Cloud Functions

La elección del tipo de proxy depende de la tarea, el presupuesto y los requisitos de anonimato. Aquí hay una comparación de las principales opciones:

Recomendaciones para la elección:

• Para scraping de marketplaces (Wildberries, Ozon, Amazon) — proxies residenciales con rotación por solicitud, para que cada solicitud provenga de una nueva IP.
• Para solicitudes API (Google Maps API, OpenWeatherMap) — proxies de centro de datos con alta velocidad, si no hay restricciones estrictas por IP.
• Para redes sociales (Instagram, TikTok) — proxies móviles, ya que tienen IP de operadores móviles y rara vez son bloqueados.
• Para scraping SEO (Google, Yandex) — proxies residenciales con geolocalización al área deseada.

Configuración de proxies en Python (requests, aiohttp)

Python es el lenguaje más popular para Cloud Functions al trabajar con scraping y automatización. Veamos la integración de proxies con las bibliotecas requests (solicitudes sincrónicas) y aiohttp (solicitudes asincrónicas).

Ejemplo con la biblioteca requests (proxy HTTP)

import requests
import os

def parse_with_proxy(request):
    # Obtenemos los datos del proxy de las variables de entorno
    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')
    
    # Formamos la URL del proxy con autenticación
    proxy_url = f"http://{proxy_user}:{proxy_pass}@{proxy_host}:{proxy_port}"
    
    proxies = {
        'http': proxy_url,
        'https': proxy_url
    }
    
    try:
        # Hacemos la solicitud a través del proxy con timeout
        response = requests.get(
            'https://api.example.com/data',
            proxies=proxies,
            timeout=10,
            headers={'User-Agent': 'Mozilla/5.0'}
        )
        
        # Verificamos el estado de la respuesta
        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'Error de proxy: {str(e)}'}
    except requests.exceptions.Timeout:
        return {'statusCode': 504, 'error': 'Timeout de solicitud'}
    except requests.exceptions.RequestException as e:
        return {'statusCode': 500, 'error': f'Error de solicitud: {str(e)}'}

Puntos importantes:

• Variables de entorno — almacene los datos del proxy (host, puerto, usuario, contraseña) en Secret Manager o variables de entorno de Cloud Functions, no en el código.
• Timeouts — asegúrese de establecer un timeout, para que la función no se bloquee en caso de problemas con el proxy.
• User-Agent — añada el encabezado User-Agent para que las solicitudes parezcan provenientes de un navegador real.
• Manejo de errores — maneje por separado ProxyError (problemas con el proxy) y Timeout (proxy lento).

Ejemplo con aiohttp (solicitudes asincrónicas)

Para tareas de alta carga (por ejemplo, scraping de más de 1000 páginas), use solicitudes asincrónicas con 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': 'Error de conexión con el proxy'}
        except asyncio.TimeoutError:
            return {'error': 'Timeout de solicitud'}

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'
    ]
    
    # Ejecutamos solicitudes asincrónicas en paralelo
    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}

El enfoque asincrónico permite realizar de 10 a 100 solicitudes paralelas a través de proxies, lo cual es crítico para el scraping de grandes volúmenes de datos dentro del tiempo de ejecución limitado de Cloud Functions (hasta 9 minutos).

Trabajo con proxies SOCKS5

Algunos proveedores de proxies ofrecen proxies SOCKS5 para un funcionamiento más confiable con tráfico UDP o eludir bloqueos. Para trabajar con SOCKS5 en Python, use la biblioteca requests[socks]:

# Añadir en 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()}

Configuración de proxies en Node.js (axios, node-fetch)

Node.js es el segundo lenguaje más popular para Cloud Functions. Veamos la integración de proxies con las bibliotecas axios y node-fetch.

Ejemplo con 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] // Devolvemos host:puerto sin contraseña
    });
    
  } catch (error) {
    if (error.code === 'ECONNREFUSED') {
      res.status(502).json({ error: 'Conexión de proxy rechazada' });
    } else if (error.code === 'ETIMEDOUT') {
      res.status(504).json({ error: 'Timeout de proxy' });
    } else {
      res.status(500).json({ error: error.message });
    }
  }
};

Dependencias para package.json:

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

Ejemplo con node-fetch y 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 });
  }
};

Dependencias para SOCKS5:

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

Autenticación de proxies: usuario/contraseña y lista blanca de IP

Existen dos métodos principales de autenticación al trabajar con proxies:

1. Autenticación por usuario y contraseña

El método más común es pasar las credenciales en la URL del proxy:

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

Ventajas: Facilidad de configuración, no requiere una IP de origen fija.

Desventajas: Las credenciales se envían en cada solicitud, pequeño overhead.

2. Autenticación por lista blanca de IP

Algunos proveedores permiten agregar direcciones IP de Cloud Functions a la lista blanca. Problema: Cloud Functions utiliza IP dinámicas del pool de Google Cloud.

Solución: Use Cloud NAT para enrutar el tráfico saliente a través de una IP externa estática:

1. Crear una red VPC y subred en Google Cloud
2. Configurar Cloud NAT con reserva de IP estática
3. Conectar Cloud Functions al VPC Connector
4. Agregar la IP estática a la lista blanca del proveedor de proxies

Después de la configuración, el proxy no requiere usuario y contraseña:

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

Recomendación: Para la mayoría de los casos, use autenticación por usuario/contraseña — es más simple y no requiere gastos adicionales en Cloud NAT (desde $0.044/hora + tráfico).

Rotación de IP y gestión de la piscina de proxies

Al hacer scraping de grandes volúmenes de datos, es crítico usar rotación de IP para evitar bloqueos. Existen varios enfoques:

1. Rotación del lado del proveedor (Rotating Proxies)

Muchos proveedores ofrecen rotating proxies — un único endpoint que cambia automáticamente la IP en cada solicitud o por temporizador:

# Un endpoint, IP cambia automáticamente
proxy_url = "http://username:password@rotating.proxy.com:8080"

# Cada solicitud proviene de una nueva IP
for i in range(100):
    response = requests.get('https://api.ipify.org', proxies={'http': proxy_url})
    print(f"Solicitud {i}: IP = {response.text}")

Ventajas: No es necesario gestionar la piscina de proxies manualmente, integración simple.

Desventajas: Sin control sobre IPs específicas, puede ser más caro.

2. Gestión manual de la piscina de proxies

Si tiene una lista de proxies estáticos, implemente la rotación a nivel de código:

import random
import requests

# Piscina de proxies (puede cargar desde 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 el proxy no funciona, probamos otro
            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 basados en sesión (sticky sessions)

Para tareas donde se necesita mantener una IP durante la sesión (por ejemplo, autenticación en un sitio), use el ID de sesión en la URL del proxy:

# Añadiendo ID de sesión al login
import uuid

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

# Todas las solicitudes con este session_id irán a través de una IP
session = requests.Session()
session.proxies = {'http': proxy_url, 'https': proxy_url}

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

# Solicitudes posteriores en la misma sesión
session.get('https://example.com/dashboard')

Manejo de errores y timeouts

Al trabajar con proxies en Cloud Functions, es crítico manejar correctamente los errores para no perder datos y no exceder los límites de tiempo de ejecución.

Tipos de errores y formas de manejo

Ejemplo de manejo integral de errores

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

def fetch_with_retry(url, proxy_pool, max_retries=3):
    """
    Solicitud con retry automático y cambio de proxy ante errores
    """
    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'}
            )
            
            # Verificamos el código de estado
            if response.status_code == 200:
                return {'success': True, 'data': response.text, 'proxy': proxy}
            elif response.status_code == 429:
                # Limitación de tasa — esperamos y probamos de nuevo
                time.sleep(2 ** attempt)  # Exponential backoff
                continue
            elif response.status_code == 403:
                # IP bloqueada — cambiamos proxy
                continue
            else:
                return {'success': False, 'status': response.status_code}
                
        except ProxyError:
            # Proxy no funciona — probamos el siguiente
            print(f"Proxy {proxy} falló, intentando otro...")
            continue
        except Timeout:
            # Timeout — probamos con otro proxy
            print(f"Timeout con {proxy}, reintentando...")
            continue
        except RequestException as e:
            # Otros errores
            print(f"La solicitud falló: {e}")
            if attempt == max_retries - 1:
                return {'success': False, 'error': str(e)}
            continue
    
    return {'success': False, 'error': 'Se superaron los máximos reintentos'}

Configuración de timeouts en Cloud Functions

Cloud Functions tienen un límite de tiempo de ejecución (por defecto 60 segundos, máximo 540 segundos). Tenga esto en cuenta al configurar los timeouts del proxy:

• Connection timeout — tiempo para establecer conexión con el proxy (se recomienda 5 segundos)
• Read timeout — tiempo para recibir respuesta del servidor objetivo a través del proxy (se recomienda 10-15 segundos)
• Total timeout — tiempo total para toda la solicitud (debe ser menor que el timeout de la función)

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

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

Mejores prácticas y optimización del rendimiento

Recomendaciones para un trabajo efectivo con proxies en Cloud Functions:

1. Use variables de entorno para credenciales

Nunca almacene nombres de usuario y contraseñas de proxies en el código. Use Secret Manager o variables de entorno:

# Creación de un secreto en Google Cloud
gcloud secrets create proxy-credentials \
    --data-file=proxy-config.json

# Proporcionar acceso a Cloud Functions
gcloud secrets add-iam-policy-binding proxy-credentials \
    --member=serviceAccount:PROJECT_ID@appspot.gserviceaccount.com \
    --role=roles/secretmanager.secretAccessor

# Lectura del secreto en el código
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. Cache los resultados del scraping

Use Cloud Storage o Firestore para almacenar en caché los datos, para no hacer solicitudes repetidas a través de proxies:

import hashlib
from google.cloud import storage

def fetch_with_cache(url, proxy):
    # Generamos una clave de caché basada en la URL
    cache_key = hashlib.md5(url.encode()).hexdigest()
    
    # Verificamos la caché en Cloud Storage
    bucket = storage.Client().bucket('my-cache-bucket')
    blob = bucket.blob(f"cache/{cache_key}.json")
    
    if blob.exists():
        # Devolvemos los datos en caché
        return json.loads(blob.download_as_text())
    
    # Hacemos la solicitud a través del proxy
    response = requests.get(url, proxies={'http': proxy})
    data = response.json()
    
    # Guardamos en caché
    blob.upload_from_string(json.dumps(data))
    
    return data

3. Monitoreo y logging

Monitoree el rendimiento de los proxies y la frecuencia de errores a través de 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. Optimización del cold start

Cloud Functions tienen un retraso de cold start. Minimice las dependencias y use versiones mínimas de bibliotecas:

# requirements.txt — solo bibliotecas necesarias
requests==2.31.0
# Evite bibliotecas pesadas como pandas, a menos que sean críticas

Use variables globales para reutilizar conexiones:

# Creamos la sesión una vez en el cold start
session = requests.Session()
session.proxies = {'http': PROXY_URL, 'https': PROXY_URL}

def parse_data(request):
    # Reutilizamos la sesión entre llamadas
    response = session.get('https://api.example.com/data')
    return response.json()

5. Geolocalización de proxies

Para tareas con geotargeting (por ejemplo, scraping de precios regionales), use proxies con vinculación a un país o ciudad específicos:

# Ejemplo con proxies residenciales, donde se puede especificar el país en el login
proxy_url = f"http://username-country-ru:password@proxy.example.com:8080"

# O usar diferentes endpoints para diferentes países
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

Conclusión

La integración de proxies con Google Cloud Functions abre amplias posibilidades para scraping, automatización y trabajo con APIs sin restricciones de IP. Los aspectos principales a considerar son: manejo adecuado de errores con lógica de reintentos, uso de timeouts para evitar bloqueos, rotación de IP para evitar bloqueos y almacenamiento seguro de credenciales en Secret Manager.

Para la mayoría de las tareas de scraping y automatización, la opción óptima son los proxies residenciales — ofrecen alta anonimidad y bajo porcentaje de bloqueos gracias al uso de IP de usuarios reales. Para trabajar con redes sociales y aplicaciones móviles, recomendamos proxies móviles, que tienen IP de operadores móviles y prácticamente no son bloqueados por plataformas como Instagram y TikTok.

Con la configuración adecuada de Cloud Functions con proxies, obtiene una solución escalable y económica para procesar grandes volúmenes de datos sin necesidad de gestionar infraestructura.