There is a moment in every scraping team's life when someone opens the scripts/ directory, sees eleven files named fetch_products_v2_FINAL.py, and realizes that the fifth scraper rewritten the same proxy rotation logic — differently, and worse — than the four before it. The code exists. The knowledge exists. It just was never packaged.
My argument: once a collection pipeline grows past a handful of targets, the proxy layer stops being a configuration concern and becomes the kernel of an internal SDK. The scripts that fetch data should be thin, declarative consumers of that kernel. Teams that make this inversion early spend their novelty budget on extraction logic and data quality; teams that don't spend it re-debugging why aiohttp raised ClientProxyConnectionError through a sticky session for the fourth time this quarter.
This article walks through the inversion concretely: what belongs in the SDK, what stays in the script, and a working skeleton you can pip install -e today.
The smell: proxy logic leaking into business code
Here is the pattern I keep seeing in the wild (and have written myself, more than once):
# fetch_products_v3.py — the 200-line "quick script"
import asyncio, aiohttp, random
PROXIES = ["http://user:pass@gate.provider.com:7000", "..."]
async def get(url):
proxy = random.choice(PROXIES) # rotation policy
for attempt in range(5): # retry policy
try:
async with aiohttp.ClientSession() as s:
async with s.get(url, proxy=proxy, timeout=aiohttp.ClientTimeout(total=30)) as r:
if r.status == 429:
await asyncio.sleep(2 ** attempt) # backoff policy
continue
return await r.text()
except aiohttp.ClientProxyConnectionError:
proxy = random.choice(PROXIES) # failure handling policy
raise RuntimeError("gave up")
Three policies are entangled with the fetch itself: rotation (which exit), retry (what to do on failure), and pacing (how hard to hit the origin). Each of those policies has proxy-specific subtleties — a 429 belongs to the exit IP that received it, not to the URL; a proxy connection error means the exit is dead, not that the site is down; a sticky session must survive retries or you lose the very cookies you were warming.
When those policies live inline, every new script re-implements them, gets them subtly wrong, and your fleet's behavior becomes the union of N private bugs.
The inversion: scripts declare, the SDK decides
The target shape is a script that reads like a spec:
from myco_collect import Collector, Target, Affinity
collector = Collector.from_env() # proxy creds, pool config, telemetry
target = Target(
name="product_detail",
affinity=Affinity.STICKY_PER_KEY, # one exit per product_id
)
async def parse(html: str) -> dict:
... # extraction logic — the only thing this script is about
for sku in ["B08N5WRWNW", "B07FZ8S74R"]:
row = await collector.fetch(
f"https://example.com/dp/{sku}",
target=target,
session_key=sku,
parser=parse,
)
Everything about how the request travels — exit selection, rotation, retries, backoff, header defaults, timeouts — is decided once, inside the SDK, where it can be tested, versioned, and observed as a unit.
A working skeleton
Here is a minimal but runnable kernel. It is deliberately small — the point is the boundaries, not the feature count.
# myco_collect/kernel.py
import asyncio
import time
import random
from dataclasses import dataclass, field
from enum import Enum
from typing import Awaitable, Callable, Optional
import aiohttp
class Affinity(Enum):
ROTATE = "rotate" # new exit per request
STICKY_PER_KEY = "sticky" # stable exit per session_key
@dataclass
class Exit:
proxy_url: str
cooldown_until: float = 0.0
failure_streak: int = 0
class ExitPool:
"""Owns exit selection. The only module that knows rotation exists."""
def __init__(self, gateway_template: str, max_sticky: int = 64):
# gateway_template e.g. "http://user:pass@gate.provider.com:7000"
self.gateway = gateway_template
self.max_sticky = max_sticky
self._sticky: dict[str, Exit] = {}
def acquire(self, affinity: Affinity, session_key: Optional[str]) -> Exit:
now = time.monotonic()
if affinity is Affinity.STICKY_PER_KEY and session_key:
exit_ = self._sticky.get(session_key)
if exit_ is None:
exit_ = Exit(self._session_gateway(session_key))
self._sticky[session_key] = exit_
if len(self._sticky) > self.max_sticky:
self._evict_coldest()
return exit_
return Exit(self.gateway) # provider rotates on its side
def _session_gateway(self, key: str) -> str:
# Many providers expose stickiness via a session token in the
# username. This is the single place that convention lives.
user, rest = self.gateway.split("://", 1)[1].split("@", 1)
user, _, password = user.partition(":")
return f"http://{user}-session-{key}:{password}@{rest}"
def _evict_coldest(self):
coldest = min(self._sticky, key=lambda k: self._sticky[k].cooldown_until)
self._sticky.pop(coldest, None)
def report(self, exit_: Exit, ok: bool) -> None:
if ok:
exit_.failure_streak = 0
else:
exit_.failure_streak += 1
if exit_.failure_streak >= 3:
exit_.cooldown_until = time.monotonic() + 120.0
@dataclass
class FetchResult:
url: str
status: int
body: str
exit_ip_hint: str
attempts: int
class Collector:
def __init__(self, pool: ExitPool, timeout_s: float = 30.0):
self.pool = pool
self.timeout = aiohttp.ClientTimeout(total=timeout_s)
@classmethod
def from_env(cls):
import os
return cls(ExitPool(os.environ["PROXY_GATEWAY"]))
async def fetch(
self,
url: str,
target,
session_key: Optional[str] = None,
parser: Optional[Callable[[str], dict]] = None,
) -> FetchResult:
attempts = 0
while True:
exit_ = self.pool.acquire(target.affinity, session_key)
if time.monotonic() < exit_.cooldown_until:
await asyncio.sleep(exit_.cooldown_until - time.monotonic())
attempts += 1
try:
connector = aiohttp.TCPConnector(limit=32)
async with aiohttp.ClientSession(
connector=connector, timeout=self.timeout
) as session:
async with session.get(url, proxy=exit_.proxy_url) as resp:
body = await resp.text()
ok = resp.status < 400 and resp.status != 429
self.pool.report(exit_, ok)
if ok:
return FetchResult(
url, resp.status, body, "gateway", attempts
)
if resp.status in (403, 429) and attempts < 4:
# penalty belongs to this exit — rotate and back off
await asyncio.sleep(min(2 ** attempts + random.random(), 30))
continue
raise RuntimeError(f"HTTP {resp.status} after {attempts} attempts")
except (aiohttp.ClientProxyConnectionError, asyncio.TimeoutError):
self.pool.report(exit_, False)
if attempts >= 4:
raise
await asyncio.sleep(0.5 * attempts)
Three details worth calling out because they are where hand-rolled scripts fail:
Stickiness is a pool concern, not a caller concern. The caller passes a session_key; the pool translates that into the provider's session convention (here, a suffix on the proxy username). When we switched providers last year, exactly one function changed.
Failures are attributed to exits, not URLs. pool.report() cools down an exit after a failure streak. A 403 means this exit is burned for this origin — retrying the same URL through the same exit is the classic rookie bug.
Retries re-enter through the pool. After a proxy-layer error, the loop calls acquire() again, so a rotating affinity gets a genuinely new exit rather than re-dialing the dead one.
What deliberately stays out
Resist the urge to make the SDK do everything. Extraction (parser= callbacks) stays in caller code, because it changes per target and per quarter, while the kernel should change per year. Persistence stays out too — the SDK returns rows; your warehouse code owns sinks. The test for the boundary is simple: if a change requires touching both the kernel and most scripts, the boundary is in the wrong place.
Packaging it so it's actually used
An internal SDK that lives in a lib/ folder is a script with delusions of grandeur. Make it a real package:
myco-collect/
pyproject.toml # name = "myco-collect", private index or git dependency
src/myco_collect/
kernel.py # the code above
py.typed # ship type hints — callers' IDEs will thank you
Then scripts depend on a version: myco-collect @ git+ssh://...@v1.4.0. That pin is what turns "which rotation logic are we running?" from an archaeology question into a one-line answer, and it means you can roll back a bad retry policy without touching a single scraper.
The payoff compounds. The sixth scraper becomes forty lines. A provider migration becomes a PR to one repo. And the person who joins in October writes extraction logic on day one instead of rediscovering, from scratch, why the fifth scraper always got 429'd at 2 a.m.
Disclosure: I use Thordata's residential proxies for this project. New users get 500MB free — code thor020 (10% off): https://www.thordata.com/?ls=uXcSHJzx&lk=02-tele
Top comments (0)