Cree un canal de búsqueda tolerante a fallos superponiendo la lógica de reintento con un retroceso exponencial, cadenas de respaldo de proveedores y un disyuntor que detenga el enrutamiento a proveedores en mal estado. Un único proveedor de búsqueda experimentará fallas transitorias: 429 límites de velocidad, 503 sobrecargas, problemas de DNS y picos de tiempo de espera. Los sistemas de producción necesitan una recuperación automática de todos estos sin perder consultas. Esta canalización maneja cada modo de falla y rastrea el estado del proveedor para que las decisiones de enrutamiento se basen en datos.
Requisitos previos
- Python 3.8+ instalado
- solicita biblioteca instalada
- Una clave API de Scavio de scavio.dev
- Comprensión de los patrones de reintento y disyuntor
Guia paso a paso
Paso 1: Definir proveedores de búsqueda
Configure proveedores primarios y alternativos con sus puntos finales y autenticación. Scavio sirve como primario con alta confiabilidad.
import os, requests, time, json
from collections import defaultdict
API_KEY = os.environ['SCAVIO_API_KEY']
PROVIDERS = [
{
'name': 'scavio',
'url': 'https://api.scavio.dev/api/v1/search',
'headers': {'x-api-key': API_KEY},
'body_fn': lambda q: {'platform': 'google', 'query': q},
'parse_fn': lambda r: r.get('organic_results', []),
},
]
health = defaultdict(lambda: {'failures': 0, 'last_failure': 0, 'circuit_open': False})Paso 2: Implementar reintento con retroceso exponencial
Envuelva cada llamada al proveedor en una lógica de reintento que maneje errores transitorios (429, 500, 502, 503) con retrasos crecientes.
def retry_search(provider: dict, query: str, max_retries: int = 3) -> list:
for attempt in range(max_retries):
try:
resp = requests.post(provider['url'],
headers=provider['headers'],
json=provider['body_fn'](query), timeout=10)
if resp.status_code == 429:
wait = 2 ** attempt
print(f"Rate limited by {provider['name']}, retrying in {wait}s")
time.sleep(wait)
continue
resp.raise_for_status()
results = provider['parse_fn'](resp.json())
health[provider['name']]['failures'] = 0
return results
except requests.exceptions.RequestException as e:
wait = 2 ** attempt
print(f"{provider['name']} attempt {attempt+1} failed: {e}, retrying in {wait}s")
time.sleep(wait)
health[provider['name']]['failures'] += 1
health[provider['name']]['last_failure'] = time.time()
return []Paso 3: Agregar cadena alternativa
Si el proveedor principal falla en todos los reintentos, pase en cascada al siguiente proveedor de la cadena. Realice un seguimiento de qué proveedor entregó el resultado.
def fallback_search(query: str) -> dict:
for provider in PROVIDERS:
name = provider['name']
if health[name]['circuit_open']:
elapsed = time.time() - health[name]['last_failure']
if elapsed < 60:
print(f'Circuit open for {name}, skipping')
continue
else:
health[name]['circuit_open'] = False
print(f'Circuit half-open for {name}, retesting')
results = retry_search(provider, query)
if results:
return {'provider': name, 'results': results}
return {'provider': 'none', 'results': []}
result = fallback_search('best crm 2026')
print(f"Served by: {result['provider']}, results: {len(result['results'])}")Paso 4: Seguimiento del estado del proveedor
Supervise las tasas de falla y abra el disyuntor cuando un proveedor supere el umbral de falla.
FAILURE_THRESHOLD = 5
def update_health(provider_name: str, success: bool):
h = health[provider_name]
if success:
h['failures'] = max(0, h['failures'] - 1)
else:
h['failures'] += 1
h['last_failure'] = time.time()
if h['failures'] >= FAILURE_THRESHOLD:
h['circuit_open'] = True
print(f'Circuit OPEN for {provider_name} ({h["failures"]} consecutive failures)')
def health_report() -> dict:
report = {}
for name, h in health.items():
report[name] = {
'failures': h['failures'],
'circuit_open': h['circuit_open'],
'last_failure': datetime.datetime.fromtimestamp(h['last_failure']).isoformat() if h['last_failure'] else 'never',
}
return report
import datetime
print(json.dumps(health_report(), indent=2))Paso 5: Probar el proceso completo
Simule fallas para verificar el comportamiento de reintento, respaldo y disyuntor de un extremo a otro.
def test_pipeline():
# Normal query
r = fallback_search('python tutorial 2026')
assert r['results'], 'Normal query should return results'
print(f'Normal: {len(r["results"])} results from {r["provider"]}')
# Multiple queries to verify consistency
queries = ['best crm', 'react vs vue', 'machine learning course']
for q in queries:
r = fallback_search(q)
print(f'{q}: {len(r["results"])} results from {r["provider"]}')
# Health check
print(f'Health: {json.dumps(health_report(), indent=2)}')
print('Pipeline test passed')
test_pipeline()Ejemplo en Python
import requests, os, time
H = {'x-api-key': os.environ['SCAVIO_API_KEY']}
def resilient_search(query, retries=3):
for i in range(retries):
try:
r = requests.post('https://api.scavio.dev/api/v1/search', headers=H,
json={'platform': 'google', 'query': query}, timeout=10)
r.raise_for_status()
return r.json().get('organic_results', [])
except Exception as e:
time.sleep(2 ** i)
return []
print(len(resilient_search('best crm 2026')), 'results')Ejemplo en JavaScript
const H = {'x-api-key': process.env.SCAVIO_API_KEY, 'Content-Type': 'application/json'};
async function resilientSearch(query, retries = 3) {
for (let i = 0; i < retries; i++) {
try {
const r = await fetch('https://api.scavio.dev/api/v1/search', {
method: 'POST', headers: H, body: JSON.stringify({platform: 'google', query})
});
if (r.ok) return (await r.json()).organic_results || [];
} catch {}
await new Promise(r => setTimeout(r, 2 ** i * 1000));
}
return [];
}
resilientSearch('best crm 2026').then(r => console.log(r.length + ' results'));Salida esperada
A search pipeline that retries transient failures with backoff, falls back across providers, and opens circuit breakers on sustained failures.