salesforce_bulk_api_insert_job_with_batch_status_polling.py

1from salesforce_bulk import SalesforceBulk
2
3# Initialize the Salesforce Bulk client
4# You will need your session ID and the instance URL (e.g., 'https://na1.salesforce.com')
5bulk = SalesforceBulk(sessionId="your_session_id", host="your_instance_url")
6
7# Define the data to be uploaded as a list of dictionaries
8data = [
9    {'FirstName': 'John', 'LastName': 'Doe', 'Email': 'john.doe@example.com'},
10    {'FirstName': 'Jane', 'LastName': 'Smith', 'Email': 'jane.smith@example.com'}
11]
12
13# Create an insert job for the 'Contact' object
14job = bulk.create_insert_job("Contact", contentType='JSON')
15
16# Add a batch of data to the job
17batch = bulk.post_batch(job, data)
18
19# Close the job
20bulk.close_job(job)
21
22# Check the status of the batch
23while not bulk.is_batch_done(batch):
24    import time
25    time.sleep(5)
26
27# Get and print the results
28results = bulk.get_all_results_for_batch(batch)
29for result in results:
30    print(result)