Salesforce REST API in Python

Jeremy Greze

The Salesforce.com REST API let one interact with the entire data of the CRM (accounts, contacts, opportunities...). An authentication is required beforehand, an access token must be obtained.

To obtain this token, we will use a connected app and an OAuth 2.0 authorization flow. The documentation indicates several available flows. In this snippet, I will use the Username-Password one, which is the easiest solution for data analysis purposes in my opinion. Alternatively, you might be interested in the Web Server and User-Agent flows that have the advantage to return also a refresh token that can be used to extend the validity of the access token.

Create an App

In any case, create a new "Connected App" in your Salesforce portal (prod or sandbox).

Go in "Setup", open the "App Manager". Then, create a "New Connected App". Name your application. Tick the box "Enable OAuth Settings". In "Selected OAuth Scopes", make all scopes available. Type "http://localhost/" in "Callback URL". Save.

At the end, you should get and note down the "Consumer Key" and the "Consumer Secret".

Use the Username-Password OAuth Flow to authenticate

With the user’s credentials and the the security token, use this Python script to get an access token.

import requests
params = {
    "grant_type": "password",
    "client_id": "XXX.YYY", # Consumer Key
    "client_secret": "0000000000000000", # Consumer Secret
    "username": "[email protected]", # The email you use to login
    "password": "MyPasswordMySecurityToken" # Concat your password and your security token
}
r = requests.post("https://login.salesforce.com/services/oauth2/token", params=params)
# if you connect to a Sandbox, use test.salesforce.com instead
access_token = r.json().get("access_token")
instance_url = r.json().get("instance_url")
print("Access Token:", access_token)
print("Instance URL", instance_url)

Make calls to the Salesforce REST API

After successfully getting an access token, one can make calls the Salesforce REST API to extract data (leads, opportunities, etc.) or to modify and create new records.

import requests

def sf_api_call(action, parameters = {}, method = 'get', data = {}):
    """
    Helper function to make calls to Salesforce REST API.
    Parameters: action (the URL), URL params, method (get, post or patch), data for POST/PATCH.
    """
    headers = {
        'Content-type': 'application/json',
        'Accept-Encoding': 'gzip',
        'Authorization': 'Bearer %s' % access_token
    }
    if method == 'get':
        r = requests.request(method, instance_url+action, headers=headers, params=parameters, timeout=30)
    elif method in ['post', 'patch']:
        r = requests.request(method, instance_url+action, headers=headers, json=data, params=parameters, timeout=10)
    else:
        # other methods not implemented in this example
        raise ValueError('Method should be get or post or patch.')
    print('Debug: API %s call: %s' % (method, r.url) )
    if r.status_code < 300:
        if method=='patch':
            return None
        else:
            return r.json()
    else:
        raise Exception('API error when calling %s : %s' % (r.url, r.content))

Very simple example: let's extract the next closing opportunities thanks to a SOQL request (Salesforce Object Query Language).

print(json.dumps(sf_api_call('/services/data/v39.0/query/', {
    'q': 'SELECT Account.Name, Name, CloseDate from Opportunity where IsClosed = False order by CloseDate ASC LIMIT 10'
}), indent=2))

The result is a JSON easy to exploit.

Voila!

Here are other examples.

Searching leads, contacts, opportunities and accounts containing a specific text

search = 'sncf'
call = sf_api_call('/services/data/v40.0/search/', parameters={
        'q': "FIND { %s } IN ALL FIELDS RETURNING Account (Id, Name), Contact (Id, Name), Opportunity (Id, Name), Lead (Id, Name) WITH METADATA='LABELS' " % search
    })
print(json.dumps(call, indent=2))

Creating an opportunity

call = sf_api_call('/services/data/v40.0/sobjects/Opportunity/', method="post", data={
    'CloseDate': '2018-03-01',
    'Name': 'My big deal',
    'StageName': 'Sales Accepted Lead',
    'Type': 'Initial Subscription',
    'AccountId': '0019E000009WTBVQA4',
})
opportunity_id = call.get('id')

Attaching a Google Doc to an opportunity

# 1) Create a ContentVersion
ContentVersion = sf_api_call('/services/data/v40.0/sobjects/ContentVersion', method="post", data={
    'Title': 'Important attached document',
    'ContentUrl': 'https://drive.google.com/drive/folders/0B60dm9TFrvQrTDVVYnp3QldvWFE'
})
ContentVersion_id = ContentVersion.get('id')

# 2) Get the ContentDocument id
ContentVersion = sf_api_call('/services/data/v40.0/sobjects/ContentVersion/%s' % ContentVersion_id)
ContentDocument_id = ContentVersion.get('ContentDocumentId')


# 3) Create a ContentDocumentLink
Opportunity_id = "0069E000003o92TQAQ"
ContentDocumentLink = sf_api_call('/services/data/v40.0/sobjects/ContentDocumentLink', method = 'post', data={
        'ContentDocumentId': ContentDocument_id,
        'LinkedEntityId': Opportunity_id,
        'ShareType': 'V'
    })

Uploading a document to an opportunity

This is very similar to the code just above. Only part 1 is modified.

# 1) Create a ContentVersion
path = "test.png"
with open(path, "rb") as f:
    encoded_string = base64.b64encode(f.read())

ContentVersion = sf_api_call('/services/data/v40.0/sobjects/ContentVersion', method="post", data={
    'Title': 'An image',
    'PathOnClient': path,
    'VersionData': encoded_string,
})
ContentVersion_id = ContentVersion.get('id')

# 2) Get the ContentDocument id (see above)

# 3) Create a ContentDocumentLink (see above)

Adding an event to an opportunity

AddEvent = sf_api_call('/services/data/v40.0/sobjects/Event', method = 'post', data={
    #'Type': 'Meeting', # seems not required
    'Subject': 'Qualification meeting',
    'OwnerId': "0050Y000000PIONQA4",
    'ActivityDate': '2017-07-24T00:00:00.000Z',
    'IsAllDayEvent': True,
    'WhoId': "0030Y00000KAABsQAP", #from doc: "represents a human such as a lead or a contact"
    'WhatId': "0060Y0000077iDGQAY", #from doc: "represents nonhuman objects such as accounts, opportunities, campaigns, cases, or custom objects"
    'Description': 'blablabla'
})

Adding a member to the OpportunityTeam of an opportunity

sf_api_call('/services/data/v40.0/sobjects/OpportunityTeamMember', method = 'post', {
    'UserId': user_id,
    'OpportunityId': opportunity_id,
    'OpportunityAccessLevel': 'Edit',
    'TeamMemberRole': 'SDR'
})

Exporting all the records of an object with all fields

For example, exporting all opportunities with all fields. You will get a list of dictionaries. You can then easily export this list to a CSV file or to a Pandas dataframe.

object_to_export = "Opportunity"
limit = 10000

describe = sf_api_call('/services/data/v40.0/sobjects/%s/describe' % object_to_export)

if not describe.get('queryable', False):
    raise Exception("This object is not queryable")

fields = [f['name'] for f in describe.get('fields')]

query = "SELECT %s FROM %s LIMIT %i" % (
    ", ".join(fields),
    object_to_export,
    limit
)

call = sf_api_call('/services/data/v40.0/queryAll/', {'q': query})

rows = call.get('records', [])
# for obj in call.get('records', []):
#     print(obj)

next = call.get('nextRecordsUrl', None)

while next:
    call = sf_api_call(next)
    rows.extend(call.get('records', []))
    # for obj in call.get('records', []):
    #     print(obj)
    next = call.get('nextRecordsUrl', None)

# print(rows)