A Salesforce CRM sync, end to end
A scheduled one-way sync from Salesforce to another system - correlation keys, change detection and a run report.
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 changed, leave all other records unchanged
- Report what happened
Service structure
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 restricts the sync to channel opportunities, which keeps both the run time and the API request spend proportional to the records the integration actually needs.
Correlating the two sides
The portal is another REST API, reached through an outgoing REST connection, and each of its deals stores 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()
# 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 single pass 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 whose values differ.
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(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(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
With the changed field names logged per update, a question such as "why did this deal's amount change on Tuesday?" can be answered from the log alone.
What a run looks like
When few records have changed, a run logs 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 - alerting is configured on a failed count other than zero, 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 take too long, filter the SOQL query on LastModifiedDate so each run reads only what changed since the previous one.
- If the business requires an auditable record of each run, extend the run report into a document - the Keysight integration renders each run's creates, updates and unmatched records into a PDF that is delivered by email.