The first integration we built was a script. Read orders from one platform, write them to a warehouse system, map the fields inline. It worked, we shipped it. The second was the same script with different field names. Still fine.
The sixth is where it turned. We had five systems talking directly to each other, and we counted the mappings one afternoon: 20 one-way translators to maintain. Adding a sixth system would take it to 30. Every new source multiplied the connections, so the cost per integration was going up, which is exactly backwards from what anyone wants by the tenth.
We routed everything through a canonical order
We stopped connecting systems to each other and connected each one to a shared internal shape.
An order from one platform and an order from another are different JSON, different field names, different nesting. But they're both an order. We defined what an order is in our world — external ID, total_cents, line items, customer email — and made every integration translate to or from that.
source A record ─┐
source B record ─┼─▶ canonical order ─┬─▶ warehouse
source C record ─┘ └─▶ analytics sink
Now N sources plus M destinations is N + M translators, not N × M. The sixth source became one inbound translator that produces canonical orders, and everything downstream already knew how to consume those. The count stopped climbing.
The math is worth being exact about, because it's the whole argument. With 5 systems all talking to each other, point-to-point, you have 5 × 4 = 20 one-way translators. Add a sixth and it's 6 × 5 = 30. The routed version at 5 systems is at most 5 + 5 = 10, and the sixth system adds 2, not 10. By the time we hit 10 systems the point-to-point version would have been 90 translators and the routed version was 20. That gap is the reason we stopped and rebuilt at six instead of pushing through to ten.
Defining the canonical order was the hard part, and we got it wrong the first time. Our first canonical order had a total field and we let each reader decide whether that meant with-tax or without-tax. Two sources disagreed, the warehouse got a mix, and the totals didn't reconcile. We split it into subtotal_cents, tax_cents, and total_cents with total = subtotal + tax as an invariant the reader has to satisfy, and validated it at the seam. The canonical model is more than field names — the definitions behind those fields are the load-bearing part, and a fuzzy definition leaks into every integration downstream of it.
Reader in, writer out, they never talk
We split every integration in half. An inbound reader that normalizes, an outbound writer that transforms. They never reference each other. They only touch the canonical model.
The reader takes a messy source record and produces a canonical one:
class ShopReader:
def normalize(self, raw: dict) -> CanonicalOrder:
return CanonicalOrder(
external_id=raw["order_number"],
total_cents=to_cents(raw["total"]),
line_items=[self._line(li) for li in raw["items"]],
)
The writer takes a canonical record and produces whatever the destination wants:
class WarehouseWriter:
def transform(self, order: CanonicalOrder) -> dict:
return {
"ref": order.external_id,
"value": from_cents(order.total_cents),
"lines": [self._line(li) for li in order.line_items],
}
Adding the analytics sink touched zero readers. Adding a third source touched zero writers. The canonical model is the seam, and the seam is the contract.
Field mapping is data, so we made it config
Most of a reader turned out to be field mapping, and field mapping is data, not logic. We pushed it into per-entity config templates:
{
"entity": "order",
"map": {
"external_id": "order_number",
"total_cents": {"from": "total", "transform": "to_cents"},
"customer_email": "buyer.email"
}
}
A generic engine reads the template and does the mapping. When we needed to pull refunds from a source we already had a reader for, it was a new template, not a new class. We only wrote code for the transforms the config couldn't express — the genuinely weird ones, a nested discount structure in one source that needed real logic. Everything routine was a JSON file an analyst could edit without touching the engine.
The dotted-path syntax handled most of the nesting. "buyer.email" walks into the nested object; the engine splits on the dot and traverses. The transforms were a small registry — to_cents, parse_date, strip_currency_symbol — named in the config and looked up at runtime:
TRANSFORMS = {
"to_cents": lambda v: round(float(v) * 100),
"parse_date": lambda v: dateutil.parse(v).isoformat(),
}
def apply_map(raw: dict, template: dict) -> dict:
out = {}
for target, rule in template["map"].items():
if isinstance(rule, str):
out[target] = dig(raw, rule) # "buyer.email"
else:
fn = TRANSFORMS[rule["transform"]]
out[target] = fn(dig(raw, rule["from"]))
return out
The one place we drew the line was arrays of arrays and conditional mapping. One source nested discounts as a list inside each line item, with a type field that changed which other fields were present. That's logic, not a field map, and forcing it into the config would have meant inventing a whole expression language inside JSON. We wrote that one reader's _line method by hand and kept the config engine simple. The rule we settled on: if expressing it in config needs branching or loops, it's code — don't grow the config format to swallow it.
Idempotent upserts and watermarks, because syncs re-run
Two invariants made this survivable in production, and both exist because syncs get re-run.
Upsert on the external ID, never plain insert. We learned this the obvious way: a crashed sync got restarted, re-processed a batch it had already written, and we had duplicate orders in the warehouse until we added the key:
db.upsert(CanonicalOrder, key="external_id", values=order.dict())
And sync incrementally. We tracked the last-processed updated_at as a watermark and pulled only what changed since:
since = store.get_watermark("shop:order")
for raw in source.fetch(updated_after=since):
...
store.set_watermark("shop:order", max_seen_updated_at)
The first version did a full re-sync every run. At a few thousand orders it was fine; at 400k it took 40 minutes and hammered the source API's rate limit. The watermark kept each run proportional to what actually changed — most runs pulled a few dozen rows. Combined with the upsert, a crashed sync resumes from the watermark and re-processing the overlap is harmless.
The watermark itself has a trap we hit once. We advanced it to now() at the end of the run instead of to the max updated_at we actually saw. A record got updated on the source during the sync, after we'd read it but before we set the watermark to now(), and the next run skipped it because its updated_at was earlier than our advanced watermark. That order silently never synced. The fix is to only ever advance the watermark to the maximum updated_at you actually processed, never to wall-clock time:
max_seen = since
for raw in source.fetch(updated_after=since):
order = reader.normalize(raw)
db.upsert(CanonicalOrder, key="external_id", values=order.dict())
max_seen = max(max_seen, raw["updated_at"])
store.set_watermark("shop:order", max_seen)
We also overlap the window by a small margin — fetch from watermark - 5 minutes — so a record that landed exactly on the boundary gets re-read. Re-reading is free because the upsert makes it idempotent. Skipping a record is not free, because you don't find out until someone notices an order missing from the warehouse weeks later.
What the shape bought
The tenth integration cost what the second did. A new source is one reader plus some config templates. A new destination is one writer plus one template. Neither touches the other half.
We paid for it up front — designing the canonical order, building the generic mapping engine — before we had enough integrations to obviously justify it. That's the uncomfortable part, and honestly we only committed to it because the sixth one hurt. But it's the only version where the cost stayed flat instead of climbing with every system we added.