Dimentica gli script usa-e-getta. Kinetic Core è un framework ingegnerizzato per ETL massivi, trasformazioni complesse e automazione DevOps.
La classe SyncPipeline non si limita a spostare dati. Gestisce batching, retry policies e
lifecycle hooks per darti il controllo totale su ogni singolo record.
from kinetic_core import SyncPipeline, SyncMode # Callback per monitoraggio real-time def on_success(record, sf_id): print(f"✓ Synced {record['email']} -> {sf_id}") # Inizializzazione Pipeline pipeline = SyncPipeline( client=client, sobject="Contact", mode=SyncMode.UPSERT, external_id_field="Ext_ID__c", batch_size=200, callbacks={ "on_record_success": on_success, "on_batch_complete": lambda b, t, r: logger.info(f"Batch {b}/{t}") } ) # Esecuzione con report finale result = pipeline.sync(source_data) print(f"Success Rate: {result.success_rate}%")
Supporto nativo per INSERT, UPDATE, DELETE e
UPSERT.
Niente logica "spaghetti" per gestire gli ID esterni: ci pensa il framework.
L'oggetto SyncResult non ti dice solo "Fatto". Ti fornisce ratei di successo,
tempo trascorso, record per secondo e lista dettagliata degli errori.
Intercetta ogni evento. Aggiorna una UI, scrivi su un DB di log o invia notifiche Slack in
tempo reale tramite i callback on_record_*.
I dati reali sono sporchi. FieldMapper ti permette di pulirli, trasformarli e
normalizzarli mentre fluiscono verso Salesforce.
Mappa strutture JSON profonde come
user.address.city direttamente su campi flat di Salesforce.
Usa qualsiasi funzione Python per trasformare i valori al volo (es. uppercase, parsing date, calcoli).
Definisci le regole di mapping in file YAML versionabili invece di hardcodarle nel codice.
# Mapping Configuration mapper = ConditionalFieldMapper( mapping={ # Simple Rename "fname": "FirstName", # Nested Access + Transform "user.contact.email": ( "Email", lambda x: x.lower() ), # Default Value "status": ( "Status__c", None, "New" ) }, # Conditional Logic (Inter-field dependency) conditions={ "Priority": lambda d: "High" if d["revenue"] > 1M else "Normal" } )
Incluso nella libreria c'è sf-toolkit, una CLI potente per operazioni rapide, debug e
ispezione della tua Org senza aprire il browser.
Gestisci lo schema della tua Org Salesforce (Campi Custom, Oggetti) direttamente da Python.
from kinetic_core.metadata import CustomField # Creo il campo "Livello AI" ai_score = CustomField( sobject="Account", name="AI_Score__c", type="Number", precision=18, scale=2, description="Punteggio generato dal modello predittivo" ) # Deploy in Org client.metadata.deploy(ai_score)
Nessun file XML da manipolare a mano. Le definizioni sono oggetti Python tipizzati.
Caricamenti massivi gestiti automaticamente. Kinetic Core si occupa dello splitting dei file CSV, del polling asincrono e dei retry.
Garantisci il futuro di Kinetic Core per il tuo business.