Enterprise CRM Integration Platform

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

class CRMConnector(ABC):
    
    @abstractmethod
    def authenticate(self, credentials: Dict) -> bool:
        """Establish connection to CRM"""
        pass
    
    @abstractmethod
    def get_schema(self) -> List[ObjectSchema]:
        """Retrieve available objects and fields"""
        pass
    
    @abstractmethod
    def read_records(self, object_type: str, 
                    since: datetime = None) -> Iterator[Record]:
        """Read records with optional delta sync"""
        pass
    
    @abstractmethod
    def write_records(self, object_type: str, 
                     records: List[Record]) -> WriteResult:
        """Create or update records"""
        pass

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

class SalesforceConnector(CRMConnector):
    
    def read_records(self, object_type: str, since: datetime = None):
        query = f"SELECT {self.fields[object_type]} FROM {object_type}"
        
        if since:
            query += f" WHERE LastModifiedDate > {since.isoformat()}"
        
        for batch in self.sf.query_all_iter(query):
            for record in batch['records']:
                yield self.normalize_record(record)

 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

class ConflictResolver:
    
    def resolve(self, records: List[ConflictingRecord], 
               rules: ConflictRules) -> Record:
        resolved = {}
        
        for field in self.get_all_fields(records):
            values = [(r.source, r.data.get(field), r.modified_at) 
                     for r in records if field in r.data]
            
            if len(set(v[1] for v in values)) == 1:
                # No conflict: all values same
                resolved[field] = values[0][1]
            else:
                # Apply resolution rule
                rule = rules.get_rule(field)
                resolved[field] = self.apply_rule(rule, values)
        
        return Record(data=resolved)
    
    def apply_rule(self, rule: str, values: List) -> Any:
        if rule == 'last_write_wins':
            return max(values, key=lambda v: v[2])[1]
        elif rule == 'source_priority':
            priority = self.source_priorities
            return min(values, key=lambda v: priority[v[0]])[1]