Building Custom Integrations: From Data Contract to Production Sync

Off-the-shelf connectors don't always fit. When you need something bespoke — a proprietary system, an unusual field mapping, an internal app with no connector — you build it yourself. This tutorial walks the full lifecycle of a custom integration: scope, design the data contract, build the sync, and operate it so it survives contact with the real world.

Table of Contents

• When You Need a Custom Integration
• Start With the Problem, Not the Code
• Define the Data Contract
• Choose the Direction of Truth
• Sync, Not Copy
• Build the Pipeline
• Make It Idempotent
• Handle the Real World: Failures
• Observe It
• Operate It: Deploy & Maintain
• A Minimal End-to-End Example

When You Need a Custom Integration

Use a custom integration when the standard tools fall short:

• The target has no connector (a legacy ERP, an internal tool).
• The field mapping is too specific for a point-and-click mapper.
• You need exact control over auth, transformations, or error handling.
• The off-the-shelf tool is too heavy or too light for the job.

Note: prefer a managed connector first if it truly fits — custom code is a maintenance commitment. Build custom only when it's genuinely needed.

Start With the Problem, Not the Code

Before writing anything, answer the "what and why" (the discipline from your Solution Design tutorials):

1. What data? Which records and fields actually matter (contact, order, stock level)?
2. Which direction? Does data flow A→B, B→A, or both?
3. How often? Real-time (webhook), near-real-time (frequent poll), or batch?
4. What's the source of truth? Which system wins when values disagree?

Best Practice: write these four answers down and keep them visible. An integration built without them drifts into feature-creep and endless field-mapping debates.

Define the Data Contract

The core artifact of a custom integration is the mapping — the translation between one system's fields and the other's. Make it explicit, in a table, and agree on it before coding:

Note: the contract (this mapping) is more durable than the code. When a field changes, you edit the mapping, not the whole integration. Keep it versioned like any other document.

Choose the Direction of Truth

Two systems will eventually disagree (a customer updated in the CRM but not the ERP). Decide who wins:

• One source of truth — one system owns a field; the other copies it. Simplest and safest.
• By-field ownership — the CRM owns email, the ERP owns price. Powerful but needs discipline.

Best Practice: for most custom integrations, make one system the source of truth and sync one-way. Bidirectional sync with conflicts is where custom integrations rot — you now need conflict resolution for every race.

Sync, Not Copy

A copy-and-paste mental model produces duplicated, drifting data. Build a sync instead: records are matched by a stable id, and each run reconciles differences rather than blindly appending.

  Match by external_id → compare fields → update what changed → never append

Your system stores the target's external_id so next time you recognize the same record. That single habit is what stops an integration from creating a duplicate on every run.

Build the Pipeline

With the contract set, the implementation is the ETL you already know (Data Engineering Tutorial 5) pointed at two systems:

import os, requests

def pull_contacts():
    """Extract: read the latest contacts from system A."""
    resp = requests.get(
        "https://api.a.com/v1/contacts",
        headers={"Authorization": f"Bearer {os.getenv('TOKEN_A')}"},
        params={"since": os.getenv("LAST_RUN")},
        timeout=30,
    )
    resp.raise_for_status()
    return resp.json()["contacts"]

def push_contact(c):
    """Load: upsert one contact into system B, matched by id."""
    resp = requests.post(
        "https://api.b.com/v1/accounts",
        headers={"Authorization": f"Bearer {os.getenv('TOKEN_B')}"},
        json={"external_id": c["id"], "name": f"{c['first']} {c['last']}"},
        timeout=30,
    )
    resp.raise_for_status()
    return resp.json()["id"]

Explanation: pull_contacts extracts with a since high-water mark (only what changed); push_contact upserts by external_id. Extract → transform (the name concat) → load, exactly as the pipeline tutorial teaches.

Make It Idempotent

Custom integrations rerun — at 3 a.m., after a crash, after you redeploy. Idempotent means a rerun produces the same result as the first run: no duplicates, no partial state. Key habits (from Data Engineering Tutorial 5):

• Match records by a stable id and upsert, never blind-append.
• Use an idempotency key on destructive POSTs so a timeout retry doesn't double-create (Tutorial 4's payment example).
• Pull by high-water mark (since) rather than a full dump each time, so reruns are cheap and bounded.

Handle the Real World: Failures

Networks fail, APIs throttle, payloads change shape. A custom integration that assumes success breaks the moment it's real. Wrap every fragile step:

import time, requests

def get_with_retry(url, headers, tries=4):
    for i in range(tries):
        r = requests.get(url, headers=headers, timeout=30)
        if r.status_code == 429:                       # rate limited
            time.sleep(int(r.headers.get("Retry-After", 2 ** i)))
            continue
        r.raise_for_status()
        return r.json()
    raise RuntimeError(f"failed after {tries} tries")

Explanation: on 429 we honor Retry-After with exponential backoff; other failures raise loudly rather than silently returning bad data (Tutorial 6 is the full treatment). A loud failure you see beats a silent wrong result you don't.

Observe It

A custom integration nobody watches quietly rots. Log the essentials and alert on anomalies:

import logging
log = logging.getLogger("integ")

def run():
    pulled = len(pull_contacts())
    pushed = sum(1 for c in pull_contacts() if push_contact(c))
    log.info("pulled=%d pushed=%d", pulled, pushed)
    if pushed == 0 and pulled > 0:
        log.warning("pulled %d but pushed 0 — check mapping", pulled)

Best Practice: log records-in vs records-out and alert when output is zero or wildly off from the rolling average. A silent integration is a failed integration — it just hasn't failed loudly yet.

Operate It: Deploy & Maintain

• Schedule with cron (simple) or an orchestrator (Airflow/Prefect) for dependencies and retries (Data Engineering Tutorial 5).
• Deploy the same hardening your other services use — secrets in a secret manager (Tutorial 2), the code in version control.
• Watch the changelogs of both APIs (Tutorial 3) — your contract is coupled to theirs. When either API changes a field, your mapping table is the first thing to revisit.
• Test any change against sandboxes before production (Tutorials 4, 6).

A Minimal End-to-End Example

Tying it together — sync contacts from A to B, idempotently, with retries:

import os, time, requests

def sync():
    headers_a = {"Authorization": f"Bearer {os.getenv('TOKEN_A')}"}
    headers_b = {"Authorization": f"Bearer {os.getenv('TOKEN_B')}"}
    pulled = get_with_retry(
        "https://api.a.com/v1/contacts?since=2026-01-01", headers_a)
    for c in pulled:
        push_contact(c)          # upsert by external_id (defined earlier)
    log.info("synced %d contacts", len(pulled))

if __name__ == "__main__":
    sync()

Explanation: extract with a since filter, push each by external_id with retry-wrapped requests, and log the count. That's a real, operable custom integration in skeleton — every piece of this tutorial is in those few lines.