No single data provider has everyone. Coverage clusters by segment, by geography, by seniority, and by how a person's email happens to be formatted. So the field you need on a given record is present in one source and missing in another, and you don't know which until you ask. On a typical B2B list, a single source returns a usable match for only 55 to 70 percent of records. The rest is the gap.
A waterfall closes the gap by asking more than one source, in order, and stopping as soon as the answer arrives. The GTM Engineer's Guide to Data Enrichment covers when a waterfall is worth the complexity and when one good provider is enough. This post is the other half: the code. If you've decided you need a waterfall, here is how to build one that merges fields correctly, resolves conflicts, caches results, and doesn't pay twice for the same lookup.
Tools like Clay made this pattern mainstream by putting dozens of providers behind one conditional-run interface, so a GTM team can chain sources without writing per-vendor glue. What follows is the version you write when you own the pipeline: a backend service, a batch job, a webhook handler. Same idea, your code. Every snippet is given in Python first, then JavaScript, so take whichever fits your stack.
What a waterfall actually computes
Strip the orchestration away and a waterfall is a loop with two rules. Ask the next provider only for the fields you still need. Stop when the record is complete or the providers run out.
The cost mechanic is the whole reason it works. Your first provider runs on every record. Your second runs only on the records the first one missed, your third only on what the second missed. Vendor benchmarks put the shape of that curve at roughly a 15 to 25 percent recovery from the second provider, 8 to 12 percent from the third, and 3 to 5 percent from a fourth (Unify, Bitscale). A single source tends to land somewhere in the 55 to 70 percent range on a typical B2B list; stacking three or four pushes that into the high eighties. The returns shrink fast, which is why most teams stop at three or four.
That curve also tells you the ordering rule. Put the cheapest, highest-hit-rate source first so the expensive long-shot only ever sees the residual. Get the order backwards and you pay premium rates on records a cheaper source would have covered.
A provider interface
Every provider returns a different JSON shape. The first thing the waterfall needs is a common one, so the merge logic doesn't care which source a field came from. Define a small adapter per provider that returns fields tagged with a confidence level and a source name.
from dataclasses import dataclass
from typing import Protocol
@dataclass
class Field:
value: str
confidence: str # "high" | "moderate" | "low"
source: str
class Provider(Protocol):
name: str
cost_units: int # relative cost of one lookup, for accounting
def lookup(self, email: str) -> dict[str, Field]:
"""Return {field_name: Field}. Missing fields are simply absent."""
...// A provider is any object: { name, costUnits, async lookup(email) }
// lookup returns { fieldName: { value, confidence, source } }
// where confidence is "high" | "moderate" | "low". Missing fields are absent.Here is the Data Legion adapter. The Person Enrich endpoint already returns per-field confidence on emails, phones, and locations, so the adapter maps that straight through instead of guessing.
import os
import requests
class DataLegionProvider:
name = "datalegion"
cost_units = 1
def lookup(self, email: str) -> dict[str, "Field"]:
resp = requests.post(
"https://api.datalegion.ai/person/enrich",
headers={"API-Key": os.environ["DATALEGION_API_KEY"]},
json={"email": email},
timeout=5,
)
if not resp.ok:
return {}
matches = resp.json().get("matches", [])
if not matches:
return {}
person = matches[0]["person"]
out: dict[str, Field] = {}
if person.get("job_title"):
out["job_title"] = Field(person["job_title"], "high", self.name)
if person.get("company_name"):
out["company"] = Field(person["company_name"], "high", self.name)
# Contact fields carry their own confidence, so use it.
for email_obj in person.get("emails", []):
if email_obj.get("type") == "professional" and email_obj.get("current"):
out["work_email"] = Field(
email_obj["address"], email_obj["confidence"], self.name
)
break
for phone in person.get("phones", []):
if phone.get("type") == "mobile" and phone.get("current"):
out["mobile_phone"] = Field(
phone["number"], phone["confidence"], self.name
)
break
return outconst dataLegion = {
name: "datalegion",
costUnits: 1,
async lookup(email) {
let resp;
try {
resp = await fetch("https://api.datalegion.ai/person/enrich", {
method: "POST",
headers: {
"API-Key": process.env.DATALEGION_API_KEY,
"Content-Type": "application/json",
},
body: JSON.stringify({ email }),
signal: AbortSignal.timeout(5000),
});
} catch {
return {};
}
if (!resp.ok) return {};
const { matches = [] } = await resp.json();
if (matches.length === 0) return {};
const person = matches[0].person;
const out = {};
const tag = (value, confidence) => ({ value, confidence, source: "datalegion" });
if (person.job_title) out.job_title = tag(person.job_title, "high");
if (person.company_name) out.company = tag(person.company_name, "high");
// Contact fields carry their own confidence, so use it.
const email_ = (person.emails || []).find(
(e) => e.type === "professional" && e.current,
);
if (email_) out.work_email = tag(email_.address, email_.confidence);
const phone = (person.phones || []).find((p) => p.type === "mobile" && p.current);
if (phone) out.mobile_phone = tag(phone.number, phone.confidence);
return out;
},
};Your other providers get the same treatment: one adapter each, all returning the same field-map shape. The waterfall never sees a vendor-specific response again.
class ProviderB:
name = "provider_b"
cost_units = 2
def lookup(self, email: str) -> dict[str, "Field"]:
# Call provider B, normalize its response into Field objects.
# Tag confidence honestly: if the provider doesn't return one,
# don't claim "high".
...const providerB = {
name: "provider_b",
costUnits: 2,
async lookup(email) {
// Call provider B, normalize its response into the field-map shape.
// Tag confidence honestly: if the provider doesn't return one,
// don't claim "high".
},
};The fall-through loop
Now the core. Walk the providers in order. Each one is asked only for the fields still missing. The moment every required field is filled, stop.
REQUIRED = ["job_title", "company", "work_email", "mobile_phone"]
def enrich(email: str, providers: list[Provider]) -> tuple[dict[str, Field], int]:
record: dict[str, Field] = {}
spent = 0
for provider in providers:
missing = [f for f in REQUIRED if f not in record]
if not missing:
break # record is complete, stop paying
spent += provider.cost_units
result = provider.lookup(email)
for name, field in result.items():
record = merge_field(record, name, field)
return record, spentconst REQUIRED = ["job_title", "company", "work_email", "mobile_phone"];
async function enrich(email, providers) {
let record = {};
let spent = 0;
for (const provider of providers) {
const missing = REQUIRED.filter((f) => !(f in record));
if (missing.length === 0) break; // record is complete, stop paying
spent += provider.costUnits;
const result = await provider.lookup(email);
for (const [name, field] of Object.entries(result)) {
record = mergeField(record, name, field);
}
}
return { record, spent };
}Two things are doing the work here. The early stop keeps you from calling provider three when provider one already returned everything. And the merge step decides what happens when a later provider returns a field you already have, which is the part most homegrown waterfalls get wrong.
Resolving conflicts
When two providers disagree, you need a rule. The naive version keeps whichever value arrived first, which means your ordering silently decides your data quality. A better rule keeps the value with the higher confidence, and only falls back to order when confidence ties.
RANK = {"high": 3, "moderate": 2, "low": 1}
def merge_field(record: dict[str, Field], name: str, incoming: Field) -> dict[str, Field]:
existing = record.get(name)
if existing is None:
record[name] = incoming
return record
# Keep the higher-confidence value. On a tie, the earlier
# provider wins, because it ran first and we trust the ordering.
if RANK[incoming.confidence] > RANK[existing.confidence]:
record[name] = incoming
return recordconst RANK = { high: 3, moderate: 2, low: 1 };
function mergeField(record, name, incoming) {
const existing = record[name];
if (!existing) {
record[name] = incoming;
return record;
}
// Keep the higher-confidence value. On a tie, the earlier
// provider wins, because it ran first and we trust the ordering.
if (RANK[incoming.confidence] > RANK[existing.confidence]) {
record[name] = incoming;
}
return record;
}This is a deliberate choice, not the only one. Some teams resolve by recency (whichever source last confirmed the value), some by a fixed provider priority per field (trust provider A for titles, provider B for phones). Pick one rule, write it down, and make it explicit in code. A waterfall without a conflict rule produces data that changes depending on which provider happened to be slow that day.
Don't pay for what you already know
A record you enriched last week shouldn't cost anything this week. Cache resolved records and check the cache before the loop runs. Even a short time-to-live saves real money on the records that get looked up repeatedly, which in most systems is a small set doing most of the volume.
import time
_cache: dict[str, tuple[dict[str, Field], float]] = {}
TTL_SECONDS = 60 * 60 * 24 * 30 # 30 days
def enrich_cached(email: str, providers: list[Provider]) -> tuple[dict[str, Field], int]:
key = email.strip().lower()
hit = _cache.get(key)
if hit and time.time() - hit[1] < TTL_SECONDS:
return hit[0], 0 # zero cost: served from cache
record, spent = enrich(key, providers)
_cache[key] = (record, time.time())
return record, spentconst cache = new Map();
const TTL_MS = 1000 * 60 * 60 * 24 * 30; // 30 days
async function enrichCached(email, providers) {
const key = email.trim().toLowerCase();
const hit = cache.get(key);
if (hit && Date.now() - hit.at < TTL_MS) {
return { record: hit.record, spent: 0 }; // zero cost: served from cache
}
const { record, spent } = await enrich(key, providers);
cache.set(key, { record, at: Date.now() });
return { record, spent };
}The in-memory store is fine for a single process. In production, move it to Redis or a table so every worker shares the same cache and a restart doesn't wipe it.
Counting what it costs
The reason to track what you spent is that it turns the diminishing-returns curve from a chart in a blog post into a number for your own data. Run a sample through the waterfall, log cost and coverage per provider, and you can see exactly where your returns flatten.
from collections import Counter
def run_batch(emails: list[str], providers: list[Provider]) -> None:
total_cost = 0
filled = Counter()
for email in emails:
record, spent = enrich_cached(email, providers)
total_cost += spent
for field in REQUIRED:
if field in record:
filled[field] += 1
n = len(emails)
print(f"records: {n} cost units: {total_cost} units/record: {total_cost / n:.2f}")
for field in REQUIRED:
print(f" {field}: {filled[field] / n:.0%} filled")async function runBatch(emails, providers) {
let totalCost = 0;
const filled = {};
for (const email of emails) {
const { record, spent } = await enrichCached(email, providers);
totalCost += spent;
for (const field of REQUIRED) {
if (field in record) filled[field] = (filled[field] || 0) + 1;
}
}
const n = emails.length;
console.log(`records: ${n} cost units: ${totalCost} units/record: ${(totalCost / n).toFixed(2)}`);
for (const field of REQUIRED) {
console.log(` ${field}: ${Math.round(((filled[field] || 0) / n) * 100)}% filled`);
}
}Order your providers cheapest-first, run a few thousand real records, and read the output. If the third provider adds two points of coverage for a third of your total spend, you have your answer about whether it belongs in the chain. The math is specific to your ICP, not the benchmark.
Coverage moves, so the job repeats
A waterfall is not a one-time fill. The people in your database keep changing jobs. LinkedIn's Work Change Report found that professionals entering the workforce today are on track to hold roughly twice as many jobs over their careers as people did fifteen years ago, and that about 70 percent of the skills used in most jobs will change by 2030. Titles move, companies change, work emails stop resolving.
So the waterfall belongs on a cadence. Re-run it against records that haven't been touched in 90 days, let the cache absorb the rest, and the cost stays proportional to how much your data has actually drifted. The real-time enrichment pipeline post covers the webhook side, enriching at the moment a record enters; the waterfall is what each of those calls fans out to once you have more than one source behind it.
Start with one provider and the merge logic above. Add a second only when you have measured the gap and the cost-per-incremental-match holds up against your numbers. The full Person Enrich field list is at www.datalegion.ai/docs/person-data/schema. Build the chain that your coverage data justifies, not the one the diagram suggests.
Build it with a coding agent
Want to scaffold this from scratch? Paste the prompt below into Claude Code, Codex, or Cursor. It carries the context the agent needs: the goal, the endpoint, the response shape, and the merge rules.
Build an enrichment waterfall that fills in a person record from an email.
Primary source: the Data Legion Person Enrich API.
- POST https://api.datalegion.ai/person/enrich
- Header: API-Key: $DATALEGION_API_KEY
- Body: {"email": "<address>"}
- Response: matches[0].person, with fields including job_title,
seniority_level, company_name, emails[], and phones[]. Each email and
phone carries a confidence of "high", "moderate", or "low". A 404 means
no match.
References for context (read these first):
- The walkthrough this is based on:
https://www.datalegion.ai/blog/post/how-to-code-an-enrichment-waterfall
- Person Enrich API: https://www.datalegion.ai/docs/api-reference/person-enrichment
- Person field schema: https://www.datalegion.ai/docs/person-data/schema
- SDKs (Python, Node): https://www.datalegion.ai/docs/sdks/python
Requirements:
- A Provider interface so I can add more sources behind the same loop.
- Call providers in priority order; stop once the fields I asked for are filled.
- Merge conflicting values by confidence (high > moderate > low).
- Cache results by normalized (lowercased, trimmed) email.
- A batch runner that stays under 100 requests per minute.
Before you start, ask me which language and framework to use. Sketch a short
plan, then build it with tests and a short README. When you finish, check the
result against each requirement above and tell me what is covered.