Salesforce API wrapper
Project description
Python Salesforce API
This project aims to provide an easy to use, highly flexible and testable solution for communicating with Salesforce through its REST and SOAP api.
Content
- Simple Usage
- Authentication
- Record management
- Quering SObjects
- Bulk
- Tooling
- Deploying
- Retrieving
- Additional features
Simple usage
Creating a new connection / client is as simple as this:
from salesforce_api import Salesforce
client = Salesforce(
username='test@example.com',
password='my-password',
security_token='password-token'
)
Authentication
To get started in the simples of ways, you would do the following
from salesforce_api import Salesforce
client = Salesforce(username='test@example.com',
password='my-password',
security_token='password-token')
If you are trying to connect to a sandbox, you have to specify this using the is_sandbox argument.
from salesforce_api import Salesforce
client = Salesforce(username='test@example.com',
password='my-password',
security_token='password-token',
is_sandbox=True)
If for some reason the login-url differs from the standard prod/test login urls, you can specify the login url. This can be useful if you are using a mock-server, for example. This will override the is_sandbox argument.
from salesforce_api import Salesforce
client = Salesforce(username='test@example.com',
password='my-password',
security_token='password-token',
domain='login.example.com')
The examples so far would use the SOAP API for authenticating. If you want to authenticate using an app, that's easy engough. The login-url and sandbox-arguments applies here as well.
from salesforce_api import Salesforce
client = Salesforce(username='test@example.com',
password='my-password',
client_id='123',
client_secret='my-secret')
If you already have an OAuth access token, obtained elsewhere, you can just as easily create a new client.
from salesforce_api import Salesforce
client = Salesforce(access_token='access-token-here',
domain='access-token-domain')
If you want to explicitly use one or the other methods of authenticating, you can do that as well
from salesforce_api import Salesforce, login
client = Salesforce(login.oauth2(username='test@example.com',
password='my-password',
client_id='123',
client_secret='my-secret'))
If you want to use a specific version of the Salesforce API, you can specify this:
from salesforce_api import Salesforce
client = Salesforce(access_token='access-token-here',
domain='access-token-domain',
api_version='51.0')
Record management
Wokring with records is easy. All SObject-related methods are exposed through the sobjects-property on the client.
The data returned from the different calls is the decoded data from the raw response.
Insert
Example
client.sobjects.Contact.insert({'LastName': 'Example', 'Email': 'test@example.com'})
Returns
{"id":"0031l000007rU3vAAE","success":true,"errors":[]}
Get
Example
client.sobjects.Contact.get('0031l000007rU3vAAE')
Returns
{
"attributes": {
"type": "Contact",
"url": "/services/data/v44.0/sobjects/Contact/0031l000007rU3vAAE"
},
"Id": "0031l000007rU3vAAE",
"LastName": "Example",
"FirstName": "Test",
...
}
Update
Example
client.sobjects.Contact.update('0031l000007rU3vAAE', {'FirstName': 'Felix', 'LastName': 'Lindstrom'})
Returns
True
Upsert
Example
client.sobjects.Contact.upsert('customExtIdField__c', '11999', {'FirstName': 'Felix', 'LastName': 'Lindstrom'})
Returns
True
Delete
Example
client.sobjects.Contact.delete('0031l000007rU3vAAE')
Returns
True
Metadata
Example
client.sobjects.Contact.metadata()
Returns
{
'objectDescribe': {
'activateable': False,
'createable': True,
'custom': False,
...
'urls': {
'compactLayouts': '/services/data/v44.0/sobjects/Contact/describe/compactLayouts',
'rowTemplate': '/services/data/v44.0/sobjects/Contact/{ID}',
'approvalLayouts': '/services/data/v44.0/sobjects/Contact/describe/approvalLayouts',
...
}
},
'recentItems': []
}
Describe
Example
client.sobjects.Contact.describe()
Returns
{
...
}
Querying SObjects
The Salesforce API is great at returning large amounts of data, so the pagination that Salesforce implements for the result of your queries is taken cared of automagically when querying for data.
Example
client.sobjects.query("SELECT Id, FirstName, LastName FROM Contact WHERE FirstName='Felix'")
Return
[{
'attributes': {
'type': 'Contact',
'url': '/services/data/v44.0/sobjects/Contact/0031l000007Jia4AAC'
},
'Id': '0031l000007Jia4AAC',
'FirstName': 'Felix',
'LastName': 'Lindstrom'
}, ...]
Bulk
This module implements the Bulk V2 API. Basically, it allows you to think less and do more.
Note that the correct permission-set might be needed on the user, see https://success.salesforce.com/issues_view?id=a1p3A000000BMPFQA4
Bulk Insert
Example
client.bulk.insert('Contact', [
{'LastName': 'Lindstrom', 'Email': 'test@example.com'},
{'LastName': 'Something else', 'Email': 'test@example.com'}
])
Returns
[<SuccessResultRecord record_id="0031l000007rU5rAAE" success="True" />,
<SuccessResultRecord record_id="0031l000007rU5sAAE" success="True" />]
Bulk Insert
Example
client.bulk.insert('Contact', [
{'LastName': 'Lindstrom', 'Email': 'test@example.com'},
{'LastName': 'Something else', 'Email': 'test@example.com'}
])
Returns
[<SuccessResultRecord record_id="0031l000007rU5rAAE" success="True" />,
<SuccessResultRecord record_id="0031l000007rU5sAAE" success="True" />]
Bulk Upsert
Example
client.bulk.upsert('Contact', [
{'LastName': 'Lindstrom', 'Email': 'test@example.com', 'MyId__c': 1},
{'LastName': 'Something else', 'Email': 'test@example.com', 'MyId__c': 2}
], external_id_field='MyId__c')
Returns
[<SuccessResultRecord record_id="0031l000007rU5rAAE" success="True" />,
<SuccessResultRecord record_id="0031l000007rU5sAAE" success="True" />]
Bulk Update
Example
client.bulk.update('Contact', [
{'LastName': 'Lindstrom', 'Email': 'test@example.com'},
{'LastName': 'Something else', 'Email': 'test@example.com'}
])
Returns
[<SuccessResultRecord record_id="0031l000007rU5rAAE" success="True" />,
<SuccessResultRecord record_id="0031l000007rU5sAAE" success="True" />]
Bulk Delete
Example
client.bulk.delete('Contact', ['0031l000007rU5rAAE', '0031l000007rU5sAAE'])
Returns
[<SuccessResultRecord record_id="0031l000007rU5rAAE" success="True" />,
<SuccessResultRecord record_id="0031l000007rU5sAAE" success="True" />]
Failed requests
Example (Given that the records no longer exists)
client.bulk.update('Contact', ['0031l000007rU5rAAE', '0031l000007rU5sAAE'])
Returns
[<FailResultRecord record_id="0031l000007rU5rAAE" success="False" error="ENTITY_IS_DELETED:entity is deleted:--" />,
<FailResultRecord record_id="0031l000007rU5sAAE" success="False" error="ENTITY_IS_DELETED:entity is deleted:--" />]
Manual managing bulk job
By using the api above, the library hides the uploading and waiting for the bulk-process to get processed.
In some cases you might want to handle this differently. Perhaps you want to upload bunch of records to be inserted and then forget about the process. This can be done by creating a job and managing it by yourself.
bulk_job = client.bulk.create_job(OPERATION.INSERT, 'Contact')
bulk_job.upload([
{'LastName': 'Lindstrom', 'Email': 'test@example.com'},
{'LastName': 'Something else', 'Email': 'test@example.com'}
])
while not bulk_job.is_done():
time.sleep(5)
Tooling
Execute Apex
Example
client.tooling.execute_apex("System.debug('Test');")
Return on success
<SuccessfulApexExecutionResult line="-1" column="-1" compiled="True" success="True" compile_problem="None" exception_stack_trace="None" exception_message="None" />
Return on failure
<FailedApexExecutionResult line="1" column="13" compiled="False" success="False" compile_problem="Unexpected token '('." exception_stack_trace="None" exception_message="None" />
Deploying
Deploying an existing package
from salesforce_api.models.deploy import Options
deployment = client.deploy.deploy('/path/to/file.zip')
deployment.wait()
result = deployment.get_status()
Only validating
from salesforce_api.models.deploy import Options
deployment = client.deploy.deploy('/path/to/file.zip', Options(checkOnly=True))
deployment.wait()
result = deployment.get_status()
Validating running specific tests
from salesforce_api.models.deploy import Options
deployment = client.deploy.deploy('/path/to/file.zip', Options(
checkOnly=True,
testLevel='RunSpecifiedTests',
runTests=[
'TesterIntegrationApplicationTest',
'TesterIntegrationProcessTest'
]
))
deployment.wait()
result = deployment.get_status()
Canceling a deployment as soon as it fails
from salesforce_api.models.deploy import Options
deployment = client.deploy.deploy('/path/to/file.zip', Options(checkOnly=True))
while not deployment.is_done():
if deployment.has_failed():
deployment.cancel()
break
Retrieving
Example
from salesforce_api.models.shared import Type
# Decide what you want to retrieve
types = [
Type('ApexTrigger'),
Type('ApexClass', ['MyMainClass', 'AnotherClass'])
]
# Create retrievement-job
retrievement = client.retrieve.retrieve(types)
# Wait for the job to finish
retrievement.wait()
# Write the resulting zip-archive to a file
open('retrieve.zip', 'wb').write(retrievement.get_zip_file().read())
Additional features
If for some reason you need to specify how the client communicates with Salesforce, you can create the requests-session yourself and pass it to the client upon creation. This would, for example, allow you proxy your requests:
import requests
from salesforce_api import Salesforce
session = requests.Session()
session.proxies.update({'https': 'https://my-proxy.com/'})
session.headers.update({'User-Agent': 'My-User-Agent'})
client = Salesforce(username='test@example.com',
password='my-password',
security_token='password-token',
session=session)
Release history Release notifications | RSS feed
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
