simple-salesforce
simple-salesforce is a low-level Python client for Salesforce REST, Bulk, Bulk 2.0, Metadata, and custom Apex APIs. It turns Salesforce objects into attributes such as sf.Contact, sends CRUD calls, and returns ordinary Python dictionaries rather than imposing an ORM. It also supplies SOQL formatting helpers, lazy query iterators, several login flows, and access to the underlying requests session. You still need to understand Salesforce objects, permissions, query languages, API limits, and authentication.
A practical, readable choice for Python code that already understands Salesforce and wants a thin synchronous client. Do not install it expecting an ORM, automatic schema knowledge, or protection from Salesforce operational limits.
Use it if
- You are writing a Python integration that needs straightforward Salesforce record CRUD and SOQL without adopting an ORM
- You need REST calls and bulk ingestion or extraction behind one authenticated client
- Your application already receives an OAuth access token and instance URL and only needs a thin API wrapper
- You need to call a custom Apex REST endpoint while keeping control of the request path and payload
- You want models, relationships, migrations, or Django QuerySets: the README explicitly describes this as a very low-level interface that returns dictionaries, so django-salesforce is the better fit
- You need the full Salesforce platform surface with generated, strongly typed service clients: this package focuses on REST, Apex, Bulk, and selected Metadata operations, not every Salesforce API
- You cannot own SOQL safety and pagination details: query strings are yours to write, quick_search performs no escaping, and query_all materializes the entire result unless you choose query_all_iter
- You want a tiny dependency tree: installation also brings requests, typing-extensions, zeep, more-itertools, and PyJWT with its cryptography extra, even if you only make a few REST calls
- You need an async client: the implementation is built on synchronous requests.Session, so network calls block unless you move them to threads or choose an async HTTP-based integration
Setup reality
Installing with pip is the easy part, but a usable connection requires Salesforce-side work. Python 3.9 or newer is required. For an existing OAuth flow, pass both the access token as session_id and the correct instance_url; the login hostname is not interchangeable with the instance returned by Salesforce. Built-in username and password login also needs either a security token, an IP-whitelisted organizationId, a JWT connected app and readable private-key file, or connected-app credentials. Sandboxes require domain='test' for the built-in login path, while My Domain values use their organization-specific prefix. Credentials, private keys, and tokens belong in a secret store, not source code. The package is synchronous and uses requests, though you can inject a configured requests.Session for proxies, custom CA bundles, retry adapters, or connection pooling. Salesforce permissions, required object fields, API-version differences, daily API quotas, and Bulk job limits are not abstracted away. SOQL remains a string language; use format_soql for values because quick_search does not escape input. query_all collects every row in memory, so large exports should use query_all_iter, Bulk lazy_operation, or Bulk 2.0 download streams. Metadata deployment also expects a correctly structured ZIP and asynchronous status polling. The installed dependencies include zeep and cryptographic JWT support, which is more machinery than the simple REST name suggests.
Patterns
Use an existing OAuth access tokenconnect-with-access-token
import os
from simple_salesforce import Salesforce
sf = Salesforce(
session_id=os.environ["SALESFORCE_ACCESS_TOKEN"],
instance_url=os.environ["SALESFORCE_INSTANCE_URL"],
)
print(sf.sf_instance)Use the instance URL returned by the OAuth flow. Do not assume the login host is the tenant's API host.
Log in with username, password, and security tokenconnect-with-security-token
import os
from simple_salesforce import Salesforce
sf = Salesforce(
username=os.environ["SALESFORCE_USERNAME"],
password=os.environ["SALESFORCE_PASSWORD"],
security_token=os.environ["SALESFORCE_SECURITY_TOKEN"],
domain="test", # remove for production login
)domain='test' selects the sandbox login endpoint and is only part of built-in authentication. Keep all three credentials out of source control.
Authenticate a connected app with JWTconnect-with-jwt
import os
from simple_salesforce import Salesforce
sf = Salesforce(
username=os.environ["SALESFORCE_USERNAME"],
consumer_key=os.environ["SALESFORCE_CONSUMER_KEY"],
privatekey_file=os.environ["SALESFORCE_PRIVATE_KEY_FILE"],
)The connected app must be configured for the JWT bearer flow, and the process must be able to read the private key file.
Manage a Salesforce recordcreate-read-update-delete
created = sf.Contact.create({
"LastName": "Lovelace",
"Email": "ada@example.com",
})
contact_id = created["id"]
contact = sf.Contact.get(contact_id)
sf.Contact.update(contact_id, {"FirstName": "Ada"})
sf.Contact.delete(contact_id)Create returns a result dictionary, while update and delete return Salesforce HTTP status codes. Required fields and permissions come from the org schema.
Bind values into a SOQL queryquery-soql-safely
from simple_salesforce import format_soql
soql = format_soql(
"SELECT Id, Email FROM Contact WHERE LastName IN {names}",
names=["Smith", "O'Reilly"],
)
result = sf.query(soql)
for record in result["records"]:
print(record["Id"], record.get("Email"))format_soql quotes and escapes values. Avoid string interpolation, and remember that quick_search explicitly performs no escaping.
Iterate through every query pagestream-query-results
for record in sf.query_all_iter(
"SELECT Id, Name FROM Account ORDER BY Id"
):
process(record)query_all builds one complete records list in memory. query_all_iter follows nextRecordsUrl lazily and is safer for large result sets.
Upsert by an external IDupsert-external-id
from simple_salesforce import format_external_id
key = format_external_id("External_Customer_ID__c", "acme/eu 42")
status = sf.Account.upsert(key, {
"Name": "Acme Europe",
"BillingCountry": "DE",
})format_external_id safely encodes values containing slashes, spaces, or other URL-sensitive characters. The Salesforce field must be marked as an External ID.
Insert records with Bulk API 1.0bulk-insert-records
rows = [
{"LastName": "Smith", "Email": "smith@example.com"},
{"LastName": "Jones", "Email": "jones@example.com"},
]
results = sf.bulk.Contact.insert(
rows, batch_size=5000, use_serial=True
)
failed = [row for row in results if not row.get("success")]The README says batches default to 10,000 and parallel concurrency. Serial mode is slower but can reduce lock contention; inspect every row result.
Download a Bulk 2.0 query in partsbulk2-query-download
query = "SELECT Id, Name, Industry FROM Account"
for part in sf.bulk2.Account.download(query, path="./exports"):
print(part)download writes result chunks to disk for lower memory use. The destination must be writable, and Bulk 2.0 jobs remain subject to Salesforce quotas.
Call a custom Apex REST endpointcall-custom-apex
payload = {"accountId": "001000000000001"}
result = sf.apexecute(
"AccountSummary",
method="POST",
data=payload,
)
print(result)The path is relative to the Apex REST base. The authenticated user still needs permission to execute the Apex class.
Inspect object fields before writingdescribe-object
description = sf.Contact.describe()
required = [
field["name"]
for field in description["fields"]
if not field["nillable"] and not field["defaultedOnCreate"]
]
print(required)Describe output reflects the connected org and API version. Field-level security can change what the current user sees.
Inject a configured requests sessionconfigure-http-session
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
session = requests.Session()
session.mount("https://", HTTPAdapter(max_retries=Retry(
total=3, backoff_factor=0.5, status_forcelist=[429, 502, 503, 504]
)))
sf = Salesforce(
session_id=token, instance_url=instance_url, session=session
)Retries can repeat writes unless the operation is idempotent. Apply a narrower policy for POST and PATCH calls, and set application-level timeouts where appropriate.
Alternatives
| Package | Registry | Pick it when |
|---|---|---|
| django-salesforce | PyPI | A Django application wants Salesforce objects exposed through models and QuerySet-style access |
| salesforce-bulk | PyPI | The job only needs the older Bulk API and a narrower client is preferable |
| zeep | PyPI | You must work directly with a Salesforce SOAP WSDL or another SOAP service |