Building Your Own Thin Wrapper Around a SERP API
Don't ship provider-specific calls across your codebase. A 100-line wrapper isolates provider quirks, makes switching trivial, and adds caching/retries in one place.
What you’ll learn
- Design a wrapper class with a stable internal interface.
- Implement provider-agnostic methods and provider-specific adapters.
- Bake in caching, retries, and normalization.
- Make provider migration a 50-line change.
A team that imports a provider's SDK directly throughout their codebase pays a high price when they switch providers. A team that wraps the provider behind a thin internal interface migrates in days.
This lesson is the wrapper pattern.
The interface
Your business logic should call methods like:
client = MySerpClient()
serp = client.search("python web scraping")
serp = client.search("pizza near me", location="Chicago,IL,United States")
It shouldn't know which provider is behind the wrapper. The wrapper exposes the methods your app needs, in your shape.
The class
# my_serp_client.py
from typing import Any, Literal
import requests, hashlib, json, time, redis
from urllib.parse import urlparse
class MySerpClient:
def __init__(self, provider: "SerpProvider", cache_ttl: int = 3600):
self.provider = provider
self.cache = redis.Redis()
self.cache_ttl = cache_ttl
def search(
self,
q: str,
gl: str = "us",
hl: str = "en",
device: Literal["desktop", "mobile"] = "desktop",
location: str | None = None,
engine: str = "google",
) -> dict:
key = self._cache_key(q, gl, hl, device, location, engine)
if cached := self.cache.get(key):
return json.loads(cached)
raw = self.provider.fetch(q=q, gl=gl, hl=hl, device=device,
location=location, engine=engine)
normalized = self.normalize(raw, engine)
self.cache.setex(key, self.cache_ttl, json.dumps(normalized))
return normalized
def normalize(self, raw: dict, engine: str) -> dict:
# Provider-specific normalization is delegated; the client
# holds the canonical OUR shape.
return {
"engine": engine,
"organic": self._extract_organic(raw),
"ads": self._extract_ads(raw),
"ai_overview": self._extract_ai(raw),
"local_pack": self._extract_local(raw),
}
@staticmethod
def _domain(url: str) -> str:
return urlparse(url).netloc.lower()
def _extract_organic(self, raw):
return [
{
"position": r.get("position"),
"title": r.get("title"),
"url": r.get("link"),
"domain": self._domain(r.get("link", "")),
"snippet": r.get("snippet"),
}
for r in raw.get("organic_results", [])
]
def _extract_ads(self, raw):
return [
{
"position": a.get("position"),
"title": a.get("title"),
"url": a.get("link"),
"block": a.get("block_position"),
}
for a in raw.get("ads", [])
]
def _extract_ai(self, raw):
ao = raw.get("ai_overview")
if not ao: return None
return {
"text": ao.get("text"),
"sources": [
{"domain": self._domain(s.get("link", "")), "rank": i + 1, "url": s.get("link")}
for i, s in enumerate(ao.get("sources", []))
],
}
def _extract_local(self, raw):
return [
{
"place_id": p.get("place_id"),
"name": p.get("title"),
"rating": p.get("rating"),
"phone": p.get("phone"),
"address": p.get("address"),
}
for p in raw.get("local_results", {}).get("places", [])
]
def _cache_key(self, q, gl, hl, device, location, engine):
s = f"{engine}|{q}|{gl}|{hl}|{device}|{location or ''}"
return f"serp:{hashlib.sha1(s.encode()).hexdigest()}"
The provider abstraction
# providers.py
import os, requests
from abc import ABC, abstractmethod
class SerpProvider(ABC):
@abstractmethod
def fetch(self, *, q, gl, hl, device, location=None, engine="google") -> dict: ...
class ProviderA(SerpProvider):
URL = "https://api.providera.com/search"
def fetch(self, *, q, gl, hl, device, location=None, engine="google"):
params = {
"q": q, "gl": gl, "hl": hl, "device": device,
"engine": engine, "api_key": os.environ["PROVIDER_A_KEY"],
}
if location: params["location"] = location
r = requests.get(self.URL, params=params, timeout=30)
r.raise_for_status()
return r.json()
class ProviderB(SerpProvider):
URL = "https://api.providerb.com/v1/serp"
def fetch(self, *, q, gl, hl, device, location=None, engine="google"):
# Provider B uses different parameter names, wrapper hides this
body = {
"search": {
"query": q,
"country": gl,
"language": hl,
"user_agent": "mobile" if device == "mobile" else "desktop",
"engine_name": engine,
}
}
if location: body["search"]["geo"] = location
r = requests.post(self.URL, json=body,
headers={"X-API-Key": os.environ["PROVIDER_B_KEY"]},
timeout=30)
r.raise_for_status()
# Provider B returns under a "data" key; normalize the SHAPE here
return r.json()["data"]
Usage
from my_serp_client import MySerpClient
from providers import ProviderA, ProviderB
# Configurable via env var
PROVIDER_NAME = os.environ.get("SERP_PROVIDER", "a")
if PROVIDER_NAME == "a":
provider = ProviderA()
else:
provider = ProviderB()
client = MySerpClient(provider)
serp = client.search("python web scraping", gl="us", hl="en")
print(serp["organic"][:3])
To switch providers: change SERP_PROVIDER env var. The rest of your codebase doesn't notice.
What goes in the wrapper vs in the app
- Wrapper: auth, retries, caching, normalization, rate limiting, provider-specific quirks.
- App: business logic, what to do with the normalized data.
If you find provider field names leaking into your app code, the wrapper is incomplete.
Provider migration drill
To prove the abstraction works, do a migration drill once:
- Sign up for Provider B's free tier.
- Implement
ProviderB.fetch. - Diff Provider B's raw response shape against Provider A's.
- Update
_extract_*methods to handle both (or branch onself.provider). - Flip the env var to B.
- Run your test suite. Should pass.
If you find your business logic needs changes, that's a sign of leakage, refactor the wrapper to absorb it.
PHP version (sketch)
<?php
interface SerpProviderInterface {
public function fetch(array $params): array;
}
class MySerpClient {
public function __construct(
private SerpProviderInterface $provider,
private \Redis $cache,
private int $cacheTtl = 3600
) {}
public function search(string $q, array $opts = []): array {
$key = $this->cacheKey($q, $opts);
if ($cached = $this->cache->get($key)) {
return json_decode($cached, true);
}
$raw = $this->provider->fetch(['q' => $q...$opts]);
$normalized = $this->normalize($raw);
$this->cache->setex($key, $this->cacheTtl, json_encode($normalized));
return $normalized;
}
// ... normalize, extract methods, cacheKey
}
Multi-provider mode
For very-resilient deployments, run multiple providers in parallel:
class FailoverProvider(SerpProvider):
def __init__(self, primary: SerpProvider, fallback: SerpProvider):
self.primary, self.fallback = primary, fallback
def fetch(self, **kw):
try:
return self.primary.fetch(**kw)
except Exception:
return self.fallback.fetch(**kw)
Wrap the wrapper. The interface stays unchanged for callers.
When NOT to wrap
If you're at hobby scale (a hundred searches/month), you don't need a wrapper. Just call the provider directly. The wrapper pays off when:
- Provider lock-in becomes a real risk.
- You want centralized caching/retries.
- Multiple parts of your codebase need SERP data.
- You may add a second provider for redundancy.
For a side project: skip it. For a real product: do it on day one.
Hands-on lab
Take the SerpClient from lesson 3.34. Refactor it into the MySerpClient + SerpProvider pattern above. Then implement a ProviderB stub (even with the same Provider A under the hood, just rename the methods). Confirm you can switch via env var. You've decoupled your business logic from any specific provider, the pattern's whole point.
Quiz, check your understanding
Pass mark is 70%. Pick the best answer; you’ll see the explanation right after.