🔗 CRM Integration

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

┌─────────────────────────────────────────────────────────────┐
│                    Your Applications                         │
│         (Website, App, Marketing, Support)                   │
└─────────────────────────────────────────────────────────────┘
                              │
                              ▼
┌─────────────────────────────────────────────────────────────┐
│                  Integration Layer                           │
│     (API Gateway, Rate Limiting, Transform, Queue)          │
└─────────────────────────────────────────────────────────────┘
                              │
        ┌─────────────────────┼─────────────────────┐
        │                     │                     │
┌───────▼───────┐   ┌─────────▼─────────┐   ┌───────▼───────┐
│  Salesforce   │   │    HubSpot        │   │  Dynamics     │
│   Adapter     │   │    Adapter        │   │   Adapter     │
└───────────────┘   └───────────────────┘   └───────────────┘

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34

# reliable sync with error handling
class CRMSync:
    def __init__(self, source: CRMAdapter, target: CRMAdapter):
        self.source = source
        self.target = target
        self.queue = TaskQueue()
    
    async def sync_all_contacts(self):
        # Fetch from source in batches
        async for batch in self.source.iter_contacts(batch_size=100):
            # Queue for processing
            await self.queue.enqueue(
                'sync_batch',
                batch,
                retry_policy=RetryPolicy(
                    max_attempts=3,
                    backoff='exponential'
                )
            )
    
    async def sync_batch(self, contacts: list[Contact]):
        results = []
        for contact in contacts:
            try:
                result = await self.target.upsert(contact)
                results.append(SyncResult.success(contact, result))
            except RateLimitError:
                # Re-queue with delay
                await self.queue.delay('sync_batch', [contact], delay=60)
            except Exception as e:
                results.append(SyncResult.failure(contact, e))
                await self.alert_on_critical(contact, e)
        
        return results

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22

class DataQualityEngine:
    def deduplicate(self, records: list[Contact]) -> list[Contact]:
        # Multiple matching strategies
        clusters = []
        
        # Exact email match
        email_groups = self.group_by_email(records)
        
        # Fuzzy name + company match
        fuzzy_groups = self.fuzzy_match(records, threshold=0.85)
        
        # Merge clusters
        for cluster in self.merge_clusters(email_groups, fuzzy_groups):
            if len(cluster) > 1:
                # Score and pick master record
                master = self.score_and_select_master(cluster)
                merged = self.merge_fields(master, cluster)
                clusters.append(merged)
            else:
                clusters.append(cluster[0])
        
        return clusters