Background
After calling serpbase, how to process the data? Three modes: streaming realtime, batch archive, cache dedup.
1. Streaming (Realtime)
import requests
import json
def stream_serp_results(query, callback):
r = requests.post(
"https://api.serpbase.dev/google/search",
headers={"X-API-Key": "sk_xxx"},
json={"q": query, "gl": "us", "num": 20},
timeout=10,
)
data = r.json()
for i, item in enumerate(data.get("organic", []), 1):
callback({
"rank": i,
"title": item["title"],
"link": item["link"],
})
def my_callback(item):
print(f" #{item['rank']}: {item['title']}")
stream_serp_results("best serp api", my_callback)
Pros: Realtime, user sees immediately
Cons: Costs high (calls API every time)
2. Batch (Archive)
import schedule
import json
from datetime import datetime
def batch_serp_processing(queries, output_file):
today = datetime.now().strftime("%Y-%m-%d")
all_results = {"date": today, "results": []}
for q in queries:
try:
r = requests.post(
"https://api.serpbase.dev/google/search",
headers={"X-API-Key": "sk_xxx"},
json={"q": q, "gl": "us", "num": 10},
timeout=10,
)
all_results["results"].append({
"query": q,
"data": r.json(),
})
except Exception as e:
print(f"{q} failed: {e}")
with open(output_file, "w") as f:
json.dump(all_results, f, ensure_ascii=False, indent=2)
schedule.every().day.at("03:00").do(
batch_serp_processing,
["SERP API", "cheap SERP API", "SERP API selection"],
"archive/2026-06-19.json"
)
Pros: Low cost (one pull all)
Cons: Data delay (24 hours)
3. Cache (Dedup)
import redis
import json
import hashlib
r = redis.Redis()
CACHE_TTL = 300 # 5 minutes
def cached_serp_search(query, gl="us"):
cache_key = f"serp:{hashlib.md5(f'{query}|{gl}'.encode()).hexdigest()}"
cached = r.get(cache_key)
if cached:
return json.loads(cached)
r = requests.post(
"https://api.serpbase.dev/google/search",
headers={"X-API-Key": "sk_xxx"},
json={"q": query, "gl": gl, "num": 5},
timeout=10,
)
data = r.json()
r.setex(cache_key, CACHE_TTL, json.dumps(data))
return data
Pros: Fast for repeated queries, cheap
Cons: First query slow, cache miss needs API
4. 3 Strategies Compared
| Dimension | Streaming | Batch | Cache |
|---|---|---|---|
| Realtime | Immediate | 24h+ delay | 5 min delay |
| Cost (1000 queries) | $0.30 | $0.30 (once daily) | $0.10 (50% hit) |
| Implementation complexity | ★ | ★★ | ★★★ |
| Best for | Realtime user queries | History archive | High-frequency repeated queries |
5. Selection
| Scenario | Recommended |
|---|---|
| Realtime user query (LLM agent / ChatBot) | Streaming + Cache |
| SEO monitoring / report | Batch (daily) |
| High-frequency query (same question repeatedly) | Cache (high hit rate) |
| Historical analysis / retrospective | Batch + Database |
6. Real Code (Combining 3 Strategies)
class SerpProcessor:
def __init__(self, cache_ttl=300):
self.cache = redis.Redis()
self.cache_ttl = cache_ttl
def get(self, query, gl="us", source="realtime"):
"""Combined entry: cache → batch → realtime"""
cached = self.cache_get(query, gl)
if cached:
return cached
if source == "batch":
return self.batch_get(query, gl)
return self.realtime_get(query, gl)
def cache_get(self, query, gl):
key = f"serp:{hash(f'{query}|{gl}')}"
cached = self.cache.get(key)
return json.loads(cached) if cached else None
def batch_get(self, query, gl):
df = pd.read_parquet("archive/*.parquet")
row = df[(df["query"] == query) & (df["gl"] == gl)].sort_values("date").iloc[-1]
return row.to_dict()
def realtime_get(self, query, gl):
r = requests.post(
"https://api.serpbase.dev/google/search",
headers={"X-API-Key": "sk_xxx"},
json={"q": query, "gl": gl, "num": 5},
timeout=10,
)
data = r.json()
key = f"serp:{hash(f'{query}|{gl}')}"
self.cache.setex(key, self.cache_ttl, json.dumps(data))
return data
7. 5 Engineering Details
Detail 1: Cache Key Design
key = f"serp:{hash(f'{query}|{gl}|{hl}|{num}')}"
Detail 2: Cache TTL
def ttl_for_query(query):
if "news" in query or "today" in query:
return 60
if "rank" in query:
return 3600
return 300
Detail 3: Batch Concurrency
import concurrent.futures
def batch_process_concurrent(queries, max_workers=10):
with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as ex:
futures = {ex.submit(self.realtime_get, q, gl): q for q in queries}
for f in concurrent.futures.as_completed(futures):
try:
yield f.result()
except Exception as e:
print(f"{futures[f]} failed: {e}")
Detail 4: Cache Breakdown Prevention
def safe_get(self, query, gl):
key = f"serp:{hash(f'{query}|{gl}')}"
cached = self.cache.get(key)
if cached:
return json.loads(cached)
lock_key = f"lock:{key}"
if not self.cache.set(lock_key, "1", ex=5, nx=True):
time.sleep(0.1)
return self.safe_get(query, gl)
try:
data = self.realtime_get(query, gl)
self.cache.setex(key, self.cache_ttl, json.dumps(data))
return data
finally:
self.cache.delete(lock_key)
Detail 5: Batch + Cache Combination
def get_with_2layer_cache(self, query, gl="us"):
if query in self.process_cache:
return self.process_cache[query]
cached = self.cache_get(query, gl)
if cached:
self.process_cache[query] = cached
return cached
data = self.realtime_get(query, gl)
self.process_cache[query] = data
return data
8. Real Data (My 1-Month Project)
| Metric | Value |
|---|---|
| Total queries | 100,000 |
| Cache hit | 60% |
| Streaming (realtime) | 40% |
| Batch | 20% |
| Cache repeat | 40% |
| serpbase monthly cost | $12 |
| Savings | 60% |
Summary
3 data processing strategies selection:
- Streaming: Realtime user query + LLM agent
- Batch: Historical archive + monitoring + report
- Cache: High-frequency repeated query + high QPS
serpbase + 3 combinations, 60% cache hit, monthly cost $12 (pure streaming $30), save 60%.
serpbase auto-refund 100% triggers, cache miss failure costs 0 credit, can confidently add multi-layer cache.
Top comments (0)