Migrate 60K Google CSE queries to a SERP API by exporting your existing query log, mapping CSE parameters to API fields, running parallel validation tests, switching production traffic, and decommissioning the CSE project. Google CSE caps at 10K queries/day on paid tiers and returns inconsistent JSON structures across schema versions. A dedicated search API removes those limits and returns normalized JSON across all query types. At $0.005/credit with Scavio, 60K monthly queries cost $300 compared to CSE's $5/1K ($300) but with structured output, multi-platform support, and no daily caps.
Prerequisites
- Python 3.8+ installed
- requests library installed
- A Scavio API key from scavio.dev
- Access to your Google CSE query logs or analytics
Walkthrough
Step 1: Export existing CSE queries
Pull your query log from CSE analytics or your application database. Group queries by frequency to prioritize migration testing.
Python
import os, json, csv
# Export from your app DB or CSE analytics CSV
def load_cse_queries(csv_path: str) -> list:
queries = []
with open(csv_path) as f:
reader = csv.DictReader(f)
for row in reader:
queries.append({
'query': row['query'],
'frequency': int(row.get('count', 1)),
'cse_params': {'cx': row.get('cx', ''), 'num': row.get('num', '10')},
})
queries.sort(key=lambda q: q['frequency'], reverse=True)
print(f'Loaded {len(queries)} unique queries ({sum(q["frequency"] for q in queries)} total calls)')
return queries
# Example with mock data:
queries = [{'query': 'best crm 2026', 'frequency': 500},
{'query': 'python async tutorial', 'frequency': 200}]
print(f'Top query: {queries[0]["query"]} ({queries[0]["frequency"]} calls/mo)')Step 2: Map CSE fields to API parameters
Map Google CSE parameters like cx, num, start, and lr to Scavio API equivalents. Most CSE params map directly or become unnecessary.
Python
import requests
API_KEY = os.environ['SCAVIO_API_KEY']
def map_cse_to_scavio(cse_query: dict) -> dict:
"""Map CSE parameters to Scavio API format."""
return {
'platform': 'google',
'query': cse_query['query'],
# CSE 'num' -> results come full by default
# CSE 'cx' -> not needed, full Google index
# CSE 'lr' -> use country/language params if needed
}
def test_single(query: dict) -> dict:
payload = map_cse_to_scavio(query)
resp = requests.post('https://api.scavio.dev/api/v1/search',
headers={'x-api-key': API_KEY},
json=payload, timeout=15)
resp.raise_for_status()
results = resp.json().get('organic_results', [])
return {'query': query['query'], 'results': len(results), 'status': 'ok'}
print(test_single({'query': 'best crm 2026'}))Step 3: Run parallel validation tests
Test the top 100 queries in parallel to validate result quality and measure latency before switching production.
Python
from concurrent.futures import ThreadPoolExecutor
import time
def validate_batch(queries: list, max_workers: int = 5) -> dict:
start = time.monotonic()
results = []
with ThreadPoolExecutor(max_workers=max_workers) as pool:
results = list(pool.map(test_single, queries[:100]))
elapsed = time.monotonic() - start
passed = sum(1 for r in results if r['results'] > 0)
failed = [r for r in results if r['results'] == 0]
print(f'Validated {len(results)} queries in {elapsed:.1f}s')
print(f'Passed: {passed}, Empty: {len(failed)}')
if failed:
print(f'Empty queries: {[f["query"] for f in failed[:5]]}')
return {'passed': passed, 'failed': len(failed), 'elapsed': round(elapsed, 1)}
validate_batch([{'query': 'best crm 2026'}, {'query': 'python async'}])Step 4: Switch production traffic
Replace the CSE HTTP call in your production code with the Scavio endpoint. Use a feature flag for gradual rollout.
Python
# Production switch with feature flag
def search(query: str, use_scavio: bool = True) -> list:
if use_scavio:
resp = requests.post('https://api.scavio.dev/api/v1/search',
headers={'x-api-key': API_KEY},
json={'platform': 'google', 'query': query}, timeout=10)
return resp.json().get('organic_results', [])
else:
# Legacy CSE call (keep as fallback during migration)
resp = requests.get('https://www.googleapis.com/customsearch/v1',
params={'key': os.environ.get('CSE_KEY', ''), 'cx': os.environ.get('CSE_CX', ''), 'q': query})
return resp.json().get('items', [])
results = search('best crm 2026', use_scavio=True)
print(f'{len(results)} results from Scavio')Step 5: Decommission the CSE project
After 7 days of stable Scavio traffic, disable CSE billing, remove CSE credentials, and archive the migration log.
Python
# Post-migration checklist:
# 1. Verify 7 days of zero CSE calls in your logs
# 2. Disable CSE API key in Google Cloud Console
# 3. Remove CSE environment variables
# 4. Archive migration validation results
def migration_report(validation: dict, daily_queries: int = 2000) -> str:
monthly = daily_queries * 30
cse_cost = monthly * 0.005 # $5/1K queries
scavio_cost = monthly * 0.005 # $0.005/credit
report = f'Migration complete: {monthly} queries/month\n'
report += f'CSE cost was: ${cse_cost:.0f}/mo\n'
report += f'Scavio cost: ${scavio_cost:.0f}/mo\n'
report += f'Validation: {validation["passed"]} passed, {validation["failed"]} empty\n'
report += 'CSE project can be decommissioned.'
return report
print(migration_report({'passed': 98, 'failed': 2}))Python Example
Python
import requests, os
H = {'x-api-key': os.environ['SCAVIO_API_KEY']}
def search(query):
data = requests.post('https://api.scavio.dev/api/v1/search', headers=H,
json={'platform': 'google', 'query': query}).json()
return data.get('organic_results', [])
# Replaces: requests.get('googleapis.com/customsearch/v1', params={...})
results = search('best crm 2026')
print(f'{len(results)} results')JavaScript Example
JavaScript
const H = {'x-api-key': process.env.SCAVIO_API_KEY, 'Content-Type': 'application/json'};
async function search(query) {
const r = await fetch('https://api.scavio.dev/api/v1/search', {
method: 'POST', headers: H, body: JSON.stringify({platform: 'google', query})
});
return (await r.json()).organic_results || [];
}
// Replaces: fetch(`googleapis.com/customsearch/v1?key=...&cx=...&q=${query}`)
search('best crm 2026').then(r => console.log(r.length + ' results'));Expected Output
JSON
A fully migrated search pipeline serving 60K+ monthly queries through Scavio instead of Google CSE, with validated result quality and a decommission checklist.