The case for a CRM abstraction layer
The naive approach to CRM integration is to write a Salesforce integration, then a HubSpot integration, then a Dynamics integration — each one a separate pile of code that speaks to a different API with a different authentication scheme, different field names, and different error formats.
This works until you have three CRMs, and then it's a maintenance nightmare. We built a framework that abstracts the differences behind a common interface — and adding a new CRM connector became a matter of days, not weeks.
The abstraction isn't magic. It's a Django model that stores CRM type, credentials, and field mappings, plus a base connector class with a defined interface: get_contacts(), upsert_contact(), get_deals(), upsert_deal(). Each CRM implements this interface. The rest of the application doesn't know which CRM it's talking to.
Authentication: every CRM does it differently
This is where CRM integrations eat time. Here's a quick map:
- Salesforce — OAuth2 with Connected App credentials, access tokens that expire, refresh tokens that don't (unless they do). Use the
simple-salesforcelibrary and handle token refresh explicitly - HubSpot — Private app tokens are the modern approach; OAuth2 for marketplace apps. Token expiry is a non-issue with private apps
- Microsoft Dynamics — Azure AD app registration, OAuth2 with client credentials flow for server-to-server. The Azure AD setup is the hard part, not the Dynamics API itself
- Pipedrive — API token per user, or OAuth2 for marketplace apps. Straightforward
- Zoho — OAuth2 with a server-based flow; tokens refresh predictably. Zoho's data centre routing (US vs EU vs India) will bite you if you don't handle it
Store credentials encrypted at rest. Rotate access tokens proactively — don't wait for a 401 in production to trigger a refresh. We use a background Celery task to refresh tokens before they expire.
Data sync: the hard problems
Bidirectional sync is a distributed systems problem dressed as a business requirement. You will encounter:
- Conflicting updates — a record updated in both systems simultaneously. Define a "source of truth" per field and enforce it; don't try to merge
- Field type mismatches — Salesforce picklist vs HubSpot enumeration vs your own enum. Maintain an explicit field mapping layer, not ad-hoc transformations scattered through the code
- Deleted records — CRMs soft-delete. Your system may hard-delete. Define what "deleted in CRM X" means for your data model before writing a line of sync code
- Webhook gaps — webhooks are delivery-unreliable. Build a polling fallback that syncs records modified in the last N minutes, even if you have webhooks configured
Rate limits: design for them from the start
Every CRM imposes rate limits. Salesforce: 15,000 API calls per 24 hours on standard licences. HubSpot: 100 requests per 10 seconds. Dynamics: varies by licence tier and often undocumented.
The pattern that works:
- Never call CRM APIs synchronously in a web request — queue the work, process it in a worker
- Implement exponential backoff with jitter on rate limit errors (HTTP 429)
- Track API call counts per CRM per period in Redis; enforce limits in the worker before hitting the API
- Use bulk APIs wherever available — Salesforce Bulk API, HubSpot batch endpoints — to reduce call volume by orders of magnitude
Testing CRM integrations without burning real API calls
CRM sandbox environments are your friend — Salesforce Developer Edition, HubSpot sandbox, Dynamics trial tenant. Set them up early and use them for integration tests.
For unit tests, mock at the HTTP level with responses or httpretty — not at the CRM client library level. This tests your serialisation logic too, not just your business logic.
import responses
@responses.activate
def test_upsert_contact_salesforce():
responses.add(
responses.POST,
"https://your-instance.salesforce.com/services/data/v58.0/sobjects/Contact/",
json={"id": "003...", "success": True},
status=201,
)
connector = SalesforceConnector(credentials=mock_credentials)
result = connector.upsert_contact({"email": "test@example.com"})
assert result.crm_id == "003..."
Surviving API changes
CRM vendors deprecate API versions. Salesforce retires old API versions annually; HubSpot has migrated its contact APIs twice in three years. To survive this:
- Pin the API version in a configuration constant, not scattered through the code
- Abstract field names behind your own canonical names — map to CRM-specific field names in the connector, not throughout the business logic
- Monitor deprecation notices via the vendor's changelog RSS or mailing list — don't find out from a production error
- Write integration tests against a live sandbox in CI so breaking changes surface before deployment
A well-structured integration layer means an API version upgrade is a connector change, not an application-wide refactor.