Python API

Salesforce

class RPA.Salesforce.Salesforce(sandbox: bool = False, domain: str = 'login')

Salesforce is a library for accessing Salesforce using REST API. The library extends simple-salesforce library.

More information available at Salesforce REST API Developer Guide.

Dataloader

The keyword execute_dataloader_import can be used to mimic Salesforce Dataloader import behaviour.

input_object can be given in different formats. Below is an example where input is in RPA.Table format in method a and list format in method b.

*** Settings ***
Library     RPA.Salesforce
Library     RPA.Database
Task Setup  Authorize Salesforce

*** Tasks ***
# Method a
${orders}=        Database Query Result As Table
...               SELECT * FROM incoming_orders
${status}=        Execute Dataloader Insert
...               ${orders}  ${mapping_dict}  Tilaus__c
# Method b
${status}=        Execute Dataloader Insert
...               ${WORKDIR}${/}orders.json  ${mapping_dict}  Tilaus__c

Example file orders.json

[
    {
        "asiakas": "0015I000002jBLIQA2"
    },
    {
        "asiakas": "0015I000002jBLDQA2"
    },
]

mapping_object describes how the input data fields are mapped into Salesforce object attributes. In the example, the mapping defines that asiakas attribute in the input object is mapped into Tilaaja__c attribute of Tilaus__c custom Salesforce object.

{
    "Tilaus__c": {
        "asiakas": "Tilaaja__c"
    },
}

Object type could be, for example, Tilaus__c.

Salesforce object operations

Following operations can be used to manage Salesforce objects:

Get Salesforce Object By Id
Create Salesforce Object
Update Salesforce Object
Upsert Salesforce Object
Delete Salesforce Object
Get Salesforce Object Metadata
Describe Salesforce Object

There are two ways to set the Salesforce domain. You can set the domain at time of library import or using the Set Domain keyword.

There are several ways to declare a domain at time of library import:

*** Settings ***
Library     RPA.Salesforce    sandbox=${TRUE}

Or using the domain to your Salesforce My domain:

*** Settings ***
Library     RPA.Salesforce    domain="robocorp"

The domain can also be set using the keyword Set Domain:

*** Settings ***
Library     RPA.Salesforce

*** Tasks ***
# Sets the domain for a sandbox environment
Set Domain    sandbox

# Sets the domain to a Salseforce My domain
Set Domain    robocorp

# Sets to domain to the default of 'login'
Set Domain

Examples

Robot Framework

*** Settings ***
Library     RPA.Salesforce
Task Setup  Authorize Salesforce

*** Variables ***
${ACCOUNT_NOKIA}    0015I000002jBLDQA2

*** Tasks ***
Change account details in Salesforce
    &{account}=      Get Salesforce Object By Id   Account  ${ACCOUNT_NOKIA}
    &{update_obj}=   Create Dictionary   Name=Nokia Ltd  BillingStreet=Nokia bulevard 1
    ${result}=       Update Salesforce Object  Account  ${ACCOUNT_NOKIA}  ${update_obj}

*** Keywords ***
Authorize Salesforce
    ${secrets}=     Get Secret   salesforce
    Auth With Token
    ...        username=${secrets}[USERNAME]
    ...        password=${secrets}[PASSWORD]
    ...        api_token=${secrets}[API_TOKEN]

Python

import pprint
from RPA.Salesforce import Salesforce
from RPA.Robocorp.Vault import FileSecrets

pp = pprint.PrettyPrinter(indent=4)
filesecrets = FileSecrets("secrets.json")
secrets = filesecrets.get_secret("salesforce")

sf = Salesforce()
sf.auth_with_token(
    username=secrets["USERNAME"],
    password=secrets["PASSWORD"],
    api_token=secrets["API_TOKEN"],
)
nokia_account_id = "0015I000002jBLDQA2"
account = sf.get_salesforce_object_by_id("Account", nokia_account_id)
pp.pprint(account)
billing_information = {
    "BillingStreet": "Nokia Bulevard 1",
    "BillingCity": "Espoo",
    "BillingPostalCode": "01210",
    "BillingCountry": "Finland",
}
result = sf.update_salesforce_object("Account", nokia_account_id, billing_information)
print(f"Update result: {result}")

ROBOT_LIBRARY_DOC_FORMAT = 'REST'

ROBOT_LIBRARY_SCOPE = 'GLOBAL'

account = {'Id': None, 'Name': None}

add_product_into_opportunity(product_name: str, quantity: int, opportunity_id: Optional[str] = None, pricebook_name: Optional[str] = None, custom_total_price: Optional[float] = None) → bool

Add Salesforce Product into Opportunity.

Parameters

product_name – type of the product in the Pricelist
quantity – number of products to add
opportunity_id – identifier of Opportunity, default None
pricebook_name – name of the pricelist, default None
custom_total_price – price that overrides quantity and product price, default None

Returns

True is operation is successful or False

auth_with_connected_app(username: str, password: str, api_token: str, consumer_key: str, consumer_secret: str, embed_api_token: bool = False) → None

Authorize to Salesforce with security token, username, password, connected app key, and connected app secret creating instance.

Parameters

username – Salesforce API username
password – Salesforce API password
api_token – Salesforce API security token
consumer_key – Salesforce connected app client ID
consumer_secret – Salesforce connected app client secret
embed_api_token – Embed API token to password (default: False)

Python

from RPA.Salesforce import Salesforce
from RPA.Robocorp.Vault import Vault

SF = Salesforce(domain="robocorp-testing-stuff.develop.my")
VAULT = Vault()

secrets = VAULT.get_secret("salesforce")
SF.auth_with_connected_app(
    username=secrets["USERNAME"],
    password=secrets["PASSWORD"],
    api_token=secrets["API_TOKEN"],
    consumer_key=secrets["CONSUMER_KEY"],
    consumer_secret=secrets["CONSUMER_SECRET"],
)

Robot Framework

*** Settings ***
Library  RPA.Salesforce   domain=robocop-testing-stuff.develop.my
Library  RPA.Robocorp.Vault

*** Tasks ***
Authenticate to Salesforce using connected app
    ${secrets}=  Get Secret  salesforce

    Auth with connected app
    ...  username=${secrets}[USERNAME]
    ...  password=${secrets}[PASSWORD]
    ...  api_token=${secrets}[API_TOKEN]
    ...  consumer_key=${secrets}[CONSUMER_KEY]
    ...  consumer_secret=${secrets}[CONSUMER_SECRET]

auth_with_token(username: str, password: str, api_token: str) → None

Authorize to Salesforce with security token, username and password creating instance.

Parameters

username – Salesforce API username
password – Salesforce API password
api_token – Salesforce API security token

create_new_opportunity(close_date: str, opportunity_name: str, stage_name: str = 'Closed Won', account_name: Optional[str] = None) → Any

Create Salesforce Opportunity object.

Parameters

close_date – closing date for the Opportunity, format ‘YYYY-MM-DD’
opportunity_name – as string
stage_name – needs to be one of the defined stages, defaults to “Closed Won”
account_name – by default uses previously set account, defaults to None

Returns

created opportunity or False

create_salesforce_object(object_type: str, object_data: Any) → dict

Create Salesforce object by type and data.

Parameters

object_type – Salesforce object type
object_data – Salesforce object data

Raises

SalesforceDataNotAnDictionary – when object_data is not dictionary

Returns

resulting object as dictionary

delete_salesforce_object(object_type: str, object_id: str) → bool

Delete Salesfoce object by type and id.

Parameters

object_type – Salesforce object type
object_id – Salesforce object id

Returns

True if successful

describe_salesforce_object(object_type: str) → dict

Get Salesfoce object description by type.

Parameters: object_type – Salesforce object type
Returns: object description as dictionary

execute_apex(apex: str, apex_data: Optional[Dict] = None, apex_method: str = 'POST', **kwargs)

Execute APEX operation.

The APEX classes can be added via Salesforce Developer console (from menu: File > New > Apex Class).

Permissions for the APEX classes can be set via Salesforce Setup (Apex Classes -> Security).

Parameters

apex – endpoint of the APEX operation
apex_data – data to be sent to the APEX operation
apex_method – operation method
kwargs – additional arguments to be passed to the APEX request

Returns

result of the APEX operation

Python

from RPA.Salesforce import Salesforce

SF = Salesforce(domain="robocorp-testing-stuff.develop.my")
# authenticate to Salesforce
SF.execute_apex(apex="MyClass", apex_data={"data": "value"})
result = SF.execute_apex(
    apex="getAccount/?id=0017R00002xmXB1QAM",
    apex_method="GET")

Robot Framework

*** Settings ***
Library  RPA.Salesforce   domain=robocop-testing-stuff.develop.my

*** Tasks ***
Executing APEX operations
    # Authenticate to Salesforce

    &{apex_data}=  Create Dictionary  data=value
    ${result}=     Execute APEX  MyClass  apex_data=${apex_data}
    ${result}=     Execute APEX
    ...  apex=getAccount/?id=0017R00002xmXB1QAM
    ...  apex_method=GET

execute_dataloader_insert(input_object: Any, mapping_object: Any, object_type: str) → bool

Keyword mimics Salesforce Dataloader ‘insert’ behaviour by taking in a input_object`representing dictionary of data to input into Salesforce, a `mapping_object representing dictionary mapping the input keys into Salesforce keys, an object_type representing Salesforce object which Datahandler will handle with operation type.

Stores operation successes into Salesforce.dataloader_success array. Stores operation errors into Salesforce.dataloader_errors.

These can be retrieved with keywords get_dataloader_success_table and get_dataloader_error_table which return corresponding data as RPA.Table.

Parameters

input_object – filepath or list of dictionaries
mapping_object – filepath or dictionary
object_type – Salesforce object type

Returns

True if operation is successful

get_dataloader_error_table() → Table: Return Dataloader error entries as RPA.Table

get_dataloader_success_table() → Table: Return Dataloader success entries as RPA.Table

get_domain() → str

Used to determine the current domain that has been set

Returns: string of the currently set domain

get_opportunity_id(opportunity_name: str) → Any

Get ID of an Opportunity linked to set account.

Parameters: opportunity_name – opportunity to query
Returns: Id of the opportunity or False

get_pricebook_entries() → dict

Get all pricebook entries.

Returns: query result

get_pricebook_id(pricebook_name: str) → Any

Get ID of a pricelist.

Returns False if unique Id is not found.

Parameters: pricebook_name – pricelist to query
Returns: Id of the pricelist or False

get_products_in_pricelist(pricebook_name: str) → dict

Get all products in a pricelist.

Parameters: pricebook_name – pricelist to query
Returns: products in dictionary

get_salesforce_object_by_id(object_type: str, object_id: str) → dict

Get Salesforce object by id and type.

Parameters

object_type – Salesforce object type
object_id – Salesforce object id

Returns

dictionary of object attributes

get_salesforce_object_metadata(object_type: str) → dict

Get Salesfoce object metadata by type.

Parameters: object_type – Salesforce object type
Returns: object metadata as dictionary

property instance

read_dictionary_from_file(mapping_file: str) → dict

Read dictionary from file.

Parameters: mapping_file – path to the file
Returns: file content as dictionary

salesforce_query(sql_string: str, as_table: bool = False) → Union[dict, Table]

Perform SQL query and return result as dict or Table.

Parameters

sql_string – SQL clause to perform.
as_table – Set to True if the result should be of RPA.Tables.Table type. (dictionary is returned by default)

Returns

Result of the SQL query.

salesforce_query_result_as_table(sql_string: str) → Table

Shorthand for Salesforce Query ${sql_string} as_table=${True}.

Parameters: sql_string – SQL clause to perform.
Returns: Result of the SQL query as RPA.Tables.Table.

property session_id

set_account(account_name: str = '', account_id: str = '') → bool

Set account name and id by giving either parameter.

Can be used together with keywords:

get_opportunity_id
create_new_opportunity

Parameters

account_name – string, defaults to “”
account_id – string, defaults to “”

Returns

True if account was found from Salesforce, else False

set_domain(domain: str = 'login') → None

Used to set the domain the Auth With Token keyword will use. To set the domain to ‘test’ or if using a sandbox environment use “sandbox” as the domain. If you have a Salsesforce My domain you may also input that name. If the domain argument is not used the default domain is “login”.

Parameters: domain – “sandbox” or the name of the Salesforce My domain; if no argument provided defaults to “login”

set_pricebook(pricebook_name: str) → None

Sets Pricebook to be used in Salesforce operations.

Parameters: pricebook_name – pricelist to use

update_salesforce_object(object_type: str, object_id: str, object_data: Any) → bool

Update Salesfoce object by type, id and data.

Parameters

object_type – Salesforce object type
object_id – Salesforce object id
object_data – Salesforce object data

Raises

SalesforceDataNotAnDictionary – when object_data is not dictionary

Returns

True if successful

upsert_salesforce_object(object_type: str, object_id: str, object_data: Any) → bool

Upsert Salesfoce object by type, id and data.

Parameters

object_type – Salesforce object type
object_id – Salesforce object id
object_data – Salesforce object data

Raises

SalesforceDataNotAnDictionary – when object_data is not dictionary

Returns

True if successful