A Salesforce CRM sync, end to end

Overview

This is the scenario most Salesforce integrations turn out to be once the demos are over: two systems each hold a copy of the same business data, one of them is the source of truth for some of the fields, and something must keep the other side current - continuously, unattended and without anyone comparing spreadsheets.

The concrete example is opportunities flowing from Salesforce to a partner portal where resellers see the deals registered for them, but the pattern is the same for products flowing to an ERP or invoices flowing from one. It is based on a real production integration running at Keysight Technologies.

Every run of the sync does five things:

• Pull the relevant records from Salesforce, with one paginated SOQL query
• Pull the corresponding records from the portal
• Correlate the two sides by a shared business key
• Write to the portal - create what is missing, update what drifted, touch nothing else
• Report what happened

The skeleton

The sync is one service, invoked by a scheduler job - every two hours, at night, or as often as the business needs. A distributed lock makes overlapping runs impossible: if a run is still going when the next tick fires, the second invocation waits.

# -*- coding: utf-8 -*-

# stdlib
from traceback import format_exc
from urllib.parse import quote

# Zato
from zato.server.service import Service

class SyncOpportunities(Service):
    name = 'crm.sync-opportunities'

    def handle(self):

        # Only one sync runs at a time, no matter how many scheduler ticks fire.
        with self.lock():

            sf_records = self.get_salesforce_opportunities()
            portal_deals = self.get_portal_deals()

            to_create, to_update = self.correlate(sf_records, portal_deals)

            created, updated, failed = self.push_changes(to_create, to_update)

            self.logger.info('Sync done, created: %s, updated: %s, failed: %s',
                created, updated, failed)

Everything below fills in the methods.

Pulling from Salesforce

One SOQL query collects the opportunities and, through relationship traversal, the partner account fields the portal needs - without traversal this would be one extra API call per record:

def get_salesforce_opportunities(self):

    conn = self.salesforce['My Salesforce Connection']

    columns = ', '.join([
        'Id', 'Name', 'StageName', 'Amount', 'CloseDate',
        'Partner_Account__r.Partner_Code__c',
        'Partner_Account__r.Name',
    ])

    query = f'SELECT {columns} FROM Opportunity ' + \
        "WHERE Channel_Or_Direct__c = 'Channel' " + \
        'ORDER BY Name ASC'

    response = conn.get('/query/?q=' + quote(query))

    page_records = response['records']
    records = list(page_records)

    # Follow the pagination trail until the result set is complete.
    while not response['done']:
        next_records_url = response['nextRecordsUrl']
        response = conn.get(next_records_url)

        page_records = response['records']
        records.extend(page_records)

    return records

The WHERE clause is doing quiet but important work - the sync reads only channel opportunities, not the whole table, which keeps both the run time and the API request spend proportional to what the integration actually cares about.

Correlating the two sides

The portal is another REST API, reached through an outgoing REST connection, and each of its deals carries the ID of the Salesforce opportunity it came from - the correlation key. Which key to use is the single most consequential design decision in a sync: it must be stable, unique and present on both sides. A Salesforce record ID stored in the target system, as here, is a good key. Names are not - they get edited, translated and duplicated.

def get_portal_deals(self):

    conn = self.rest['Partner Portal']
    response = conn.get(self.cid)

    # Index the deals by their correlation key - the Salesforce opportunity ID.
    deals = {}

    for deal in response.data:
        crm_opportunity_id = deal['crm_opportunity_id']
        deals[crm_opportunity_id] = deal

    return deals

With both sides in hand, correlation is a walk over the Salesforce records. Each one either has a counterpart in the portal or does not - and for the counterparts, only those whose fields actually differ become updates:

def correlate(self, sf_records, portal_deals):

    to_create = []
    to_update = []

    for record in sf_records:

        opportunity_id = record['Id']
        partner_account = record['Partner_Account__r']

        # What this deal should look like in the portal
        desired = {
            'name': record['Name'],
            'stage': record['StageName'],
            'amount': record['Amount'],
            'partner_code': partner_account['Partner_Code__c'],
        }

        # No counterpart - the whole record is to be created ..
        if not (deal := portal_deals.get(opportunity_id)):
            to_create.append((opportunity_id, desired))
            continue

        # .. otherwise, collect only the fields that drifted.
        changes = {}

        for field_name, new_value in desired.items():
            if deal[field_name] != new_value:
                changes[field_name] = new_value

        if changes:
            to_update.append((deal['id'], changes))

    return to_create, to_update

Note what the code does not do: it never touches portal deals that have no Salesforce counterpart. Those may be manual entries, records from another source or leftovers under investigation - a one-way sync that deletes whatever it cannot match is how integrations end up in incident reports. If unmatched records are a problem, report them and let a person decide.

Writing the changes

Each write is caught individually, so one rejected record does not take down the rest of the run - the policy discussed in the error handling guide:

def push_changes(self, to_create, to_update):

    conn = self.rest['Partner Portal']

    created = 0
    updated = 0
    failed  = 0

    for opportunity_id, deal in to_create:
        try:
            request = {'crm_opportunity_id': opportunity_id, **deal}
            _ = conn.post(self.cid, request)
            created += 1
        except Exception:
            failed += 1
            self.logger.warning('Could not create `%s`, e:`%s`', opportunity_id, format_exc())

    for deal_id, changes in to_update:
        try:
            _ = conn.patch(self.cid, changes, params={'id': deal_id})
            updated += 1

            changed_fields = sorted(changes)
            self.logger.info('Updated deal %s, fields: %s', deal_id, changed_fields)
        except Exception:
            failed += 1
            self.logger.warning('Could not update `%s`, e:`%s`', deal_id, format_exc())

    return created, updated, failed

Logging the changed field names per update is worth the line it costs - a question like "why did this deal's amount change on Tuesday?" is answered by the log, not by an archaeology session.

What a run looks like

On a quiet afternoon, the whole story is three lines:

INFO - crm.sync-opportunities - Collected 412 opportunities from Salesforce
INFO - crm.sync-opportunities - Updated deal 88123, fields: ['amount', 'stage']
INFO - crm.sync-opportunities - Sync done, created: 0, updated: 1, failed: 0

Four hundred records read, one field-level change detected, one write made. That last line is also the integration's health signal - a failed count that stops being zero is where alerting hooks in, and the /limits/ endpoint logged alongside it tracks what the sync costs the org.

Where to take it further

• If the other system also changes data that must flow back into Salesforce, run a second sync in the opposite direction, writing with upserts by external ID so replays stay harmless - two one-way syncs with clear field ownership beat one "bidirectional" sync every time.
• If runs grow long, filter the SOQL query on LastModifiedDate so each run reads only what changed since the previous one.
• If the business wants proof, extend the run report into a document - the Keysight integration renders each run's creates, updates and unmatched records into a PDF that goes out by email.

Learn more