tessl/pypi-simple-salesforce
A basic Salesforce.com REST API client for Python applications.
—
Pending
rest-api.mddocs/
REST API Operations
Core Salesforce class and SFType interface providing complete CRUD operations, queries, searches, and metadata access for Salesforce objects through the REST API. These components form the foundation for all standard Salesforce interactions.
Core Salesforce Class
The primary client class managing authentication, session state, and providing access to all Salesforce REST API endpoints.
class Salesforce:
def __init__(
self,
username=None,
password=None,
security_token=None,
session_id=None,
instance=None,
instance_url=None,
organizationId=None,
version="59.0",
proxies=None,
session=None,
client_id=None,
domain=None,
consumer_key=None,
consumer_secret=None,
privatekey_file=None,
privatekey=None,
parse_float=None,
object_pairs_hook=dict
):
"""
Initialize Salesforce REST API client.
Parameters:
- Authentication parameters: See authentication.md for details
- version: Salesforce API version (default: "59.0")
- proxies: HTTP proxy configuration dictionary
- session: Custom requests.Session object
- parse_float: Custom JSON float parser function
- object_pairs_hook: JSON object parsing hook (default: dict)
"""Key Properties
Session and endpoint information available after successful authentication:
class Salesforce:
@property
def session_id(self) -> str:
"""Current Salesforce session ID"""
@property
def sf_instance(self) -> str:
"""Salesforce instance domain (e.g., 'na1.salesforce.com')"""
@property
def sf_version(self) -> str:
"""API version in use"""
@property
def base_url(self) -> str:
"""Base REST API URL"""
@property
def bulk_url(self) -> str:
"""Bulk API v1.0 URL"""
@property
def bulk2_url(self) -> str:
"""Bulk API v2.0 URL"""
@property
def metadata_url(self) -> str:
"""Metadata API URL"""
@property
def tooling_url(self) -> str:
"""Tooling API URL"""
@property
def oauth2_url(self) -> str:
"""OAuth 2.0 API URL"""
@property
def api_usage(self) -> dict:
"""API usage statistics from last request"""
@property
def mdapi(self):
"""Metadata API interface (SfdcMetadataApi instance)"""Query Operations
SOQL query execution with support for large result sets and deleted records.
class Salesforce:
def query(self, query, include_deleted=False, **kwargs):
"""
Execute SOQL query and return results.
Parameters:
- query: SOQL query string
- include_deleted: Include deleted/archived records
- **kwargs: Additional request parameters (headers, etc.)
Returns:
dict: Query results with 'records', 'totalSize', 'done', and 'nextRecordsUrl'
"""
def query_more(self, next_records_identifier, identifier_is_url=False, include_deleted=False, **kwargs):
"""
Retrieve additional query results using nextRecordsUrl or identifier.
Parameters:
- next_records_identifier: Next records URL or identifier
- identifier_is_url: True if identifier is complete URL
- include_deleted: Include deleted/archived records
- **kwargs: Additional request parameters
Returns:
dict: Additional query results
"""
def query_all(self, query, include_deleted=False, **kwargs):
"""
Execute query and automatically retrieve all results.
Parameters:
- query: SOQL query string
- include_deleted: Include deleted/archived records
- **kwargs: Additional request parameters
Returns:
dict: Complete query results with all records
"""
def query_all_iter(self, query, include_deleted=False, **kwargs):
"""
Lazy iterator for query results, yielding records as retrieved.
Parameters:
- query: SOQL query string
- include_deleted: Include deleted/archived records
- **kwargs: Additional request parameters
Yields:
dict: Individual record dictionaries
"""Query Usage Examples
from simple_salesforce import Salesforce
sf = Salesforce(username='user@example.com', password='pass', security_token='token')
# Basic query
results = sf.query("SELECT Id, Name FROM Account LIMIT 10")
for record in results['records']:
print(f"Account: {record['Name']}")
# Query with deleted records
deleted_results = sf.query(
"SELECT Id, Name FROM Account WHERE IsDeleted = true",
include_deleted=True
)
# Large result set with pagination
all_results = sf.query_all("SELECT Id, Name FROM Account")
print(f"Total accounts: {all_results['totalSize']}")
# Memory-efficient iteration over large results
for record in sf.query_all_iter("SELECT Id, Name FROM Account"):
process_account(record)Search Operations
SOSL search functionality for cross-object text searching.
class Salesforce:
def search(self, search):
"""
Execute SOSL search with complete search string.
Parameters:
- search: Complete SOSL search string including FIND clause
Returns:
list: Search results grouped by object type
"""
def quick_search(self, search):
"""
Execute simplified SOSL search (auto-wraps in FIND clause).
Parameters:
- search: Search terms (automatically wrapped in FIND{})
Returns:
list: Search results grouped by object type
"""Search Usage Examples
# Full SOSL search
search_results = sf.search(
"FIND {test} IN ALL FIELDS RETURNING Account(Id, Name), Contact(Id, Name)"
)
# Quick search (simplified)
quick_results = sf.quick_search("test company")
# Process search results
for sobject_results in search_results:
print(f"Object type: {sobject_results['attributes']['type']}")
for record in sobject_results['records']:
print(f" {record['Name']}")Describe Operations
Metadata and schema information retrieval for objects and the organization.
class Salesforce:
def describe(self, **kwargs):
"""
Describe all available Salesforce objects in the organization.
Parameters:
- **kwargs: Additional request parameters (headers, etc.)
Returns:
dict: Organization describe information with 'sobjects' list
"""
def is_sandbox(self):
"""
Check if the current organization is a sandbox.
Returns:
bool: True if sandbox, False if production
"""Describe Usage Examples
# Get all available objects
org_describe = sf.describe()
for sobject in org_describe['sobjects']:
if sobject['createable']:
print(f"Createable object: {sobject['name']}")
# Check environment
if sf.is_sandbox():
print("Connected to sandbox environment")
else:
print("Connected to production environment")Direct REST Operations
Low-level REST API access for custom endpoints and advanced operations.
class Salesforce:
def restful(self, path, params=None, method='GET', **kwargs):
"""
Make direct REST API calls to Salesforce endpoints.
Parameters:
- path: API endpoint path (relative to base URL)
- params: Query parameters dictionary
- method: HTTP method (GET, POST, PUT, PATCH, DELETE)
- **kwargs: Additional request parameters (data, headers, etc.)
Returns:
requests.Response: Raw HTTP response object
"""
def oauth2(self, path, params=None, method='GET'):
"""
Make calls to OAuth 2.0 API endpoints.
Parameters:
- path: OAuth endpoint path
- params: Query parameters dictionary
- method: HTTP method
Returns:
requests.Response: Raw HTTP response object
"""Direct REST Usage Examples
# Custom REST endpoint
response = sf.restful('sobjects/Account/describe')
account_metadata = response.json()
# OAuth introspection
token_info = sf.oauth2('introspect', params={'token': sf.session_id})
# Custom API endpoint with POST
custom_response = sf.restful(
'sobjects/Account',
method='POST',
json={'Name': 'Test Account', 'Type': 'Customer'}
)Specialized API Operations
Tooling and Apex APIs
class Salesforce:
def toolingexecute(self, action, method='GET', data=None, **kwargs):
"""
Execute Tooling API operations for development tools.
Parameters:
- action: Tooling API endpoint path
- method: HTTP method
- data: Request body data
- **kwargs: Additional request parameters
Returns:
dict: Tooling API response
"""
def apexecute(self, action, method='GET', data=None, **kwargs):
"""
Execute Apex REST API operations for custom Apex endpoints.
Parameters:
- action: Apex REST endpoint path
- method: HTTP method
- data: Request body data
- **kwargs: Additional request parameters
Returns:
dict: Apex REST response
"""User Management
class Salesforce:
def set_password(self, user, password):
"""
Set password for a Salesforce user.
Parameters:
- user: User ID (18-character Salesforce ID)
- password: New password string
Returns:
dict: Password reset response
"""Organization Information
class Salesforce:
def limits(self, **kwargs):
"""
Get organization limits and usage information.
Parameters:
- **kwargs: Additional request parameters
Returns:
dict: Organization limits including API usage, storage, etc.
"""Dynamic SObject Access
The Salesforce class provides dynamic attribute access to create SFType instances for any Salesforce object.
sf = Salesforce(username='user@example.com', password='pass', security_token='token')
# Access any standard or custom object
accounts = sf.Account # Returns SFType for Account
contacts = sf.Contact # Returns SFType for Contact
custom = sf.MyCustom__c # Returns SFType for custom object
# Access bulk handlers
bulk_v1 = sf.bulk # Returns SFBulkHandler
bulk_v2 = sf.bulk2 # Returns SFBulk2HandlerSFType Class
Interface for operations on specific Salesforce SObject types, providing CRUD operations, metadata access, and specialized functionality.
class SFType:
def __init__(
self,
object_name,
session_id,
sf_instance,
sf_version="59.0",
proxies=None,
session=None,
salesforce=None,
parse_float=None,
object_pairs_hook=dict
):
"""
Initialize SFType for specific Salesforce object.
Parameters:
- object_name: Salesforce object API name
- session_id: Authenticated session ID
- sf_instance: Salesforce instance URL
- sf_version: API version string
- proxies: HTTP proxy configuration
- session: Custom requests.Session object
- salesforce: Parent Salesforce instance
- parse_float: Custom JSON float parser
- object_pairs_hook: JSON object parsing hook
"""SFType Properties
class SFType:
@property
def name(self) -> str:
"""SObject API name"""
@property
def session_id(self) -> str:
"""Session ID for authentication"""
@property
def base_url(self) -> str:
"""Base URL for this SObject type"""
@property
def api_usage(self) -> dict:
"""API usage statistics from last request"""CRUD Operations
Complete Create, Read, Update, Delete operations for Salesforce records.
class SFType:
def create(self, data, headers=None):
"""
Create new record for this SObject type.
Parameters:
- data: Record field data dictionary
- headers: Optional HTTP headers
Returns:
dict: Created record information with 'id' and 'success'
"""
def get(self, record_id, headers=None, **kwargs):
"""
Get record by Salesforce ID.
Parameters:
- record_id: 15 or 18-character Salesforce record ID
- headers: Optional HTTP headers
- **kwargs: Additional query parameters
Returns:
dict: Complete record data
"""
def get_by_custom_id(self, custom_id_field, custom_id, headers=None, **kwargs):
"""
Get record by external/custom ID field.
Parameters:
- custom_id_field: External ID field API name
- custom_id: External ID value
- headers: Optional HTTP headers
- **kwargs: Additional query parameters
Returns:
dict: Complete record data
"""
def update(self, record_id, data, raw_response=False, headers=None):
"""
Update existing record.
Parameters:
- record_id: Salesforce record ID to update
- data: Updated field data dictionary
- raw_response: Return raw HTTP response if True
- headers: Optional HTTP headers
Returns:
int|dict: HTTP status code (204) or error details if raw_response=True
"""
def upsert(self, record_id, data, raw_response=False, headers=None):
"""
Create or update record (upsert operation).
Parameters:
- record_id: External ID value for upsert matching
- data: Record field data dictionary
- raw_response: Return raw HTTP response if True
- headers: Optional HTTP headers
Returns:
dict: Upsert result with 'id', 'success', and 'created' status
"""
def delete(self, record_id, raw_response=False, headers=None):
"""
Delete record (soft delete).
Parameters:
- record_id: Salesforce record ID to delete
- raw_response: Return raw HTTP response if True
- headers: Optional HTTP headers
Returns:
int|dict: HTTP status code (204) or error details if raw_response=True
"""Metadata Operations
Schema and metadata information for SObject types.
class SFType:
def metadata(self, headers=None):
"""
Get SObject metadata information.
Parameters:
- headers: Optional HTTP headers
Returns:
dict: Complete SObject metadata
"""
def describe(self, headers=None):
"""
Get detailed SObject describe information including fields.
Parameters:
- headers: Optional HTTP headers
Returns:
dict: SObject describe with fields, record types, etc.
"""
def describe_layout(self, record_id, headers=None):
"""
Get layout information for a specific record.
Parameters:
- record_id: Record ID for layout context
- headers: Optional HTTP headers
Returns:
dict: Layout information for the record
"""Change Tracking
Track record changes over time periods for data synchronization.
class SFType:
def updated(self, start, end, headers=None):
"""
Get list of updated records in date range.
Parameters:
- start: Start date (ISO format or datetime)
- end: End date (ISO format or datetime)
- headers: Optional HTTP headers
Returns:
dict: List of record IDs updated in the time period
"""
def deleted(self, start, end, headers=None):
"""
Get list of deleted records in date range.
Parameters:
- start: Start date (ISO format or datetime)
- end: End date (ISO format or datetime)
- headers: Optional HTTP headers
Returns:
dict: List of record IDs deleted in the time period
"""File Operations
Handle base64-encoded file uploads and downloads for file fields.
class SFType:
def upload_base64(self, file_path, base64_field='Body', headers=None, **kwargs):
"""
Upload file as base64-encoded content to create new record.
Parameters:
- file_path: Path to local file for upload
- base64_field: Field name for base64 content (default: 'Body')
- headers: Optional HTTP headers
- **kwargs: Additional record field data
Returns:
dict: Created record information
"""
def update_base64(self, record_id, file_path, base64_field='Body', headers=None, raw_response=False, **kwargs):
"""
Update record with base64-encoded file content.
Parameters:
- record_id: Record ID to update
- file_path: Path to local file for upload
- base64_field: Field name for base64 content
- headers: Optional HTTP headers
- raw_response: Return raw HTTP response if True
- **kwargs: Additional field updates
Returns:
int|dict: HTTP status or response data
"""
def get_base64(self, record_id, base64_field='Body', data=None, headers=None, **kwargs):
"""
Get base64-encoded file content from record.
Parameters:
- record_id: Record ID containing file data
- base64_field: Field name containing base64 content
- data: Optional additional query data
- headers: Optional HTTP headers
- **kwargs: Additional query parameters
Returns:
dict: Record data including base64 file content
"""SFType Usage Examples
from simple_salesforce import Salesforce
sf = Salesforce(username='user@example.com', password='pass', security_token='token')
# Get SFType for Account object
account_type = sf.Account
# Create new account
new_account = account_type.create({
'Name': 'Test Account',
'Type': 'Customer',
'Industry': 'Technology'
})
print(f"Created account ID: {new_account['id']}")
# Get account by ID
account = account_type.get(new_account['id'])
print(f"Account name: {account['Name']}")
# Update account
account_type.update(new_account['id'], {
'Phone': '555-123-4567',
'Website': 'https://example.com'
})
# Get account metadata
metadata = account_type.describe()
for field in metadata['fields']:
if field['type'] == 'email':
print(f"Email field: {field['name']}")
# Check for recent changes
from datetime import datetime, timedelta
end_date = datetime.now()
start_date = end_date - timedelta(days=7)
updated_records = account_type.updated(start_date, end_date)
print(f"Updated records: {updated_records}")
# File upload example (for Document or Attachment)
document_type = sf.Document
doc_result = document_type.upload_base64(
'/path/to/file.pdf',
Name='Important Document',
FolderId='00l000000000000' # Document folder ID
)Utility Methods
Helper functions for response processing and API usage analysis.
class Salesforce:
def parse_result_to_json(self, result):
"""
Parse requests.Response object to JSON with error handling.
Parameters:
- result: requests.Response object
Returns:
dict: Parsed JSON response data
"""
@staticmethod
def parse_api_usage(sforce_limit_info):
"""
Parse Salesforce API usage information from response headers.
Parameters:
- sforce_limit_info: Sforce-Limit-Info header value
Returns:
dict: Parsed API usage statistics
"""Install with Tessl CLI
npx tessl i tessl/pypi-simple-salesforce