Forget throw-away scripts. Kinetic Core is an engineered framework for Massive ETL, Complex Transformations, and DevOps Automation.
The SyncPipeline class doesn't just move data. It handles batching, retry policies, and
lifecycle hooks to give you total control over every single record.
from kinetic_core import SyncPipeline, SyncMode # Real-time monitoring callback def on_success(record, sf_id): print(f"✓ Synced {record['email']} -> {sf_id}") # Pipeline Initialization 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}") } ) # Execute with final report result = pipeline.sync(source_data) print(f"Success Rate: {result.success_rate}%")
Native support for INSERT, UPDATE, DELETE, and
UPSERT.
No "spaghetti" logic to manage external IDs: the framework handles it.
The SyncResult object doesn't just say "Done". It provides success rates,
elapsed time, records per second, and a detailed list of errors.
Intercept every event. Update a UI, write to a log DB, or send Slack notifications in
real-time via on_record_* callbacks.
Real data is messy. FieldMapper allows you to clean, transform, and normalize it as it
flows into Salesforce.
Map deep JSON structures like user.address.city
directly to flat Salesforce fields.
Use any Python function to transform values on the fly (e.g., uppercase, date parsing, calculations).
Define mapping rules in versionable YAML files instead of hardcoding them.
# 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" } )
Included in the library is sf-toolkit, a powerful CLI for quick operations, debug, and org
inspection without opening a browser.
Manage your Salesforce Org schema (Custom Fields, Objects) directly from Python.
from kinetic_core.metadata import CustomField # Create "AI Level" field ai_score = CustomField( sobject="Account", name="AI_Score__c", type="Number", precision=18, scale=2, description="Score generated by predictive model" ) # Deploy to Org client.metadata.deploy(ai_score)
No XML files to hand-edit. Definitions are typed Python objects.
Mass uploads handled automatically. Kinetic Core handles CSV splitting, asynchronous polling, and retries.
Every single class is documented with real-world examples.
Read the DocsSecure the future of Kinetic Core for your business.