tessl/pypi-simple-salesforce
A basic Salesforce.com REST API client for Python applications.
—
Pending
exceptions.mddocs/
Exception Classes and Error Handling
Comprehensive exception hierarchy mapping HTTP status codes and Salesforce errors to specific Python exceptions for precise error handling and debugging. The simple-salesforce library provides detailed error information to help developers identify and resolve issues quickly.
Base Exception Class
The foundation exception class that all other Salesforce-specific exceptions inherit from.
class SalesforceError(Exception):
def __init__(self, url, status, resource_name, content):
"""
Base exception for all Salesforce API errors.
Parameters:
- url: API endpoint URL where error occurred
- status: HTTP status code
- resource_name: Salesforce resource or operation name
- content: Response content with error details
Attributes:
- url: The API endpoint URL
- status: HTTP status code
- resource_name: Resource identifier
- content: Raw error response content
"""
self.url = url
self.status = status
self.resource_name = resource_name
self.content = content
# Create detailed error message
message = f"Error Code: {status}"
if resource_name:
message += f", Resource: {resource_name}"
if url:
message += f", URL: {url}"
super().__init__(message)HTTP Status Code Exceptions
Specific exceptions mapped to HTTP status codes for precise error handling.
class SalesforceMoreThanOneRecord(SalesforceError):
"""
Error Code 300: Multiple Choice
Raised when a query or operation returns multiple records but only one was expected.
Common in get operations with non-unique identifiers.
"""
class SalesforceMalformedRequest(SalesforceError):
"""
Error Code 400: Bad Request
Raised when the request is malformed, has invalid parameters, or contains
invalid data. Check request syntax, required fields, and data formats.
"""
class SalesforceExpiredSession(SalesforceError):
"""
Error Code 401: Unauthorized
Raised when the session has expired or authentication credentials are invalid.
Requires re-authentication or session refresh.
"""
class SalesforceRefusedRequest(SalesforceError):
"""
Error Code 403: Forbidden
Raised when the request is refused due to insufficient permissions.
The user may lack required object or field-level permissions.
"""
class SalesforceResourceNotFound(SalesforceError):
"""
Error Code 404: Not Found
Raised when the requested resource, record, or endpoint does not exist.
Verify record IDs, object names, and API endpoints.
"""Authentication and General Exceptions
Specialized exceptions for authentication failures and general Salesforce errors.
class SalesforceAuthenticationFailed(SalesforceError):
"""
Authentication failure exception.
Raised during login when credentials are invalid, security tokens are wrong,
or authentication methods are not properly configured.
"""
class SalesforceGeneralError(SalesforceError):
"""
General Salesforce error for unspecified issues.
Raised for Salesforce-specific errors that don't fit other categories,
such as org limits, configuration issues, or service unavailability.
"""Bulk API v2.0 Exceptions
Specialized exceptions for Bulk API v2.0 operations with enhanced error details.
class SalesforceOperationError(SalesforceError):
"""
Base exception for Bulk API v2.0 operation errors.
Provides additional context for bulk operation failures including
job information and processing details.
"""
class SalesforceBulkV2LoadError(SalesforceOperationError):
"""
Bulk API v2.0 data load/ingest error.
Raised when bulk insert, update, upsert, or delete operations fail.
Contains details about failed records and validation errors.
"""
class SalesforceBulkV2ExtractError(SalesforceOperationError):
"""
Bulk API v2.0 data extraction/query error.
Raised when bulk query operations fail due to query syntax errors,
resource limitations, or data access issues.
"""Exception Handling Patterns
Basic Exception Handling
from simple_salesforce import (
Salesforce,
SalesforceAuthenticationFailed,
SalesforceExpiredSession,
SalesforceResourceNotFound,
SalesforceMalformedRequest,
SalesforceRefusedRequest,
SalesforceError
)
try:
sf = Salesforce(
username='user@example.com',
password='password',
security_token='token'
)
# Perform operations
account = sf.Account.get('001XX000003DHPr')
except SalesforceAuthenticationFailed as e:
print(f"Authentication failed: {e}")
print("Check username, password, and security token")
except SalesforceExpiredSession as e:
print(f"Session expired: {e}")
print("Re-authenticate to get new session")
except SalesforceResourceNotFound as e:
print(f"Resource not found: {e}")
print("Verify the record ID exists")
except SalesforceMalformedRequest as e:
print(f"Bad request: {e}")
print("Check request parameters and data format")
except SalesforceRefusedRequest as e:
print(f"Access denied: {e}")
print("Check user permissions for this operation")
except SalesforceError as e:
print(f"General Salesforce error: {e}")
print(f"Status: {e.status}")
print(f"URL: {e.url}")
print(f"Content: {e.content}")Advanced Exception Analysis
def analyze_salesforce_error(error):
"""
Analyze Salesforce error and provide detailed diagnostics.
Parameters:
- error: SalesforceError exception instance
Returns:
dict: Detailed error analysis and suggested actions
"""
analysis = {
'error_type': type(error).__name__,
'status_code': getattr(error, 'status', None),
'url': getattr(error, 'url', None),
'resource': getattr(error, 'resource_name', None),
'content': getattr(error, 'content', None),
'suggestions': []
}
# Parse error content for additional details
if hasattr(error, 'content') and error.content:
try:
import json
if isinstance(error.content, str):
content_data = json.loads(error.content)
else:
content_data = error.content
if isinstance(content_data, list) and len(content_data) > 0:
error_detail = content_data[0]
analysis['error_code'] = error_detail.get('errorCode')
analysis['message'] = error_detail.get('message')
analysis['fields'] = error_detail.get('fields', [])
except:
# Content is not JSON or not parseable
pass
# Add specific suggestions based on error type
if isinstance(error, SalesforceAuthenticationFailed):
analysis['suggestions'].extend([
"Verify username and password are correct",
"Check if security token is required and valid",
"Ensure the user account is not locked or deactivated",
"Verify the correct Salesforce domain (login vs test)"
])
elif isinstance(error, SalesforceExpiredSession):
analysis['suggestions'].extend([
"Re-authenticate to obtain a fresh session",
"Check session timeout settings",
"Implement session refresh logic for long-running processes"
])
elif isinstance(error, SalesforceResourceNotFound):
analysis['suggestions'].extend([
"Verify the record ID is valid and exists",
"Check if the object type is correct",
"Ensure the record hasn't been deleted",
"Verify API endpoint URLs are correct"
])
elif isinstance(error, SalesforceMalformedRequest):
analysis['suggestions'].extend([
"Check required fields are provided",
"Verify field names match the object schema",
"Ensure data types match field requirements",
"Check for invalid characters or formatting"
])
elif isinstance(error, SalesforceRefusedRequest):
analysis['suggestions'].extend([
"Check user has appropriate object permissions",
"Verify field-level security settings",
"Check if user has required profile or permission sets",
"Verify organization-wide defaults allow access"
])
return analysis
# Usage
try:
result = sf.Account.create({
'Name': '', # Empty required field
'InvalidField__c': 'value' # Non-existent field
})
except SalesforceError as e:
error_analysis = analyze_salesforce_error(e)
print(f"Error Type: {error_analysis['error_type']}")
print(f"Status: {error_analysis['status_code']}")
if 'message' in error_analysis:
print(f"Message: {error_analysis['message']}")
if error_analysis['suggestions']:
print("Suggestions:")
for suggestion in error_analysis['suggestions']:
print(f" - {suggestion}")Bulk Operations Error Handling
def handle_bulk_errors(bulk_results, operation_type="bulk operation"):
"""
Handle and analyze bulk operation errors.
Parameters:
- bulk_results: Results from bulk operation
- operation_type: Description of the operation for logging
Returns:
dict: Error analysis and failed record details
"""
successful_records = []
failed_records = []
error_summary = {}
for i, result in enumerate(bulk_results):
if result.get('success', False):
successful_records.append(i)
else:
failed_records.append({
'index': i,
'error': result.get('error', 'Unknown error'),
'id': result.get('id'),
'result': result
})
# Categorize errors
error_msg = result.get('error', 'Unknown')
error_type = error_msg.split(':')[0] if ':' in error_msg else error_msg
error_summary[error_type] = error_summary.get(error_type, 0) + 1
success_rate = (len(successful_records) / len(bulk_results)) * 100 if bulk_results else 0
print(f"{operation_type.title()} Results:")
print(f" Total records: {len(bulk_results)}")
print(f" Successful: {len(successful_records)}")
print(f" Failed: {len(failed_records)}")
print(f" Success rate: {success_rate:.1f}%")
if error_summary:
print(" Error breakdown:")
for error_type, count in error_summary.items():
print(f" {error_type}: {count}")
return {
'success_count': len(successful_records),
'failure_count': len(failed_records),
'success_rate': success_rate,
'failed_records': failed_records,
'error_summary': error_summary
}
# Usage with bulk operations
try:
bulk_data = [
{'Name': 'Valid Account', 'Type': 'Customer'},
{'Name': '', 'Type': 'Customer'}, # Will fail - empty Name
{'Name': 'Another Valid Account', 'InvalidField': 'value'} # Will fail - invalid field
]
results = sf.bulk.Account.insert(bulk_data, include_detailed_results=True)
error_analysis = handle_bulk_errors(results, "bulk insert")
# Handle failed records
if error_analysis['failed_records']:
print("\nFailed record details:")
for failed in error_analysis['failed_records']:
print(f" Record {failed['index']}: {failed['error']}")
except SalesforceError as e:
print(f"Bulk operation failed entirely: {e}")Retry Logic with Exception Handling
import time
import random
def retry_with_backoff(operation_func, max_retries=3, base_delay=1, max_delay=60):
"""
Retry operations with exponential backoff and exception handling.
Parameters:
- operation_func: Function to retry
- max_retries: Maximum number of retry attempts
- base_delay: Initial delay between retries
- max_delay: Maximum delay between retries
Returns:
Result of successful operation
Raises:
Last exception if all retries fail
"""
last_exception = None
for attempt in range(max_retries + 1):
try:
return operation_func()
except SalesforceExpiredSession as e:
print(f"Attempt {attempt + 1}: Session expired, re-authenticating...")
# Re-authentication would happen here
last_exception = e
except SalesforceRefusedRequest as e:
# Don't retry permission errors
print(f"Permission denied, not retrying: {e}")
raise
except SalesforceResourceNotFound as e:
# Don't retry for missing resources
print(f"Resource not found, not retrying: {e}")
raise
except (SalesforceMalformedRequest, SalesforceGeneralError) as e:
print(f"Attempt {attempt + 1}: Transient error, retrying... {e}")
last_exception = e
except SalesforceError as e:
print(f"Attempt {attempt + 1}: Unknown Salesforce error: {e}")
last_exception = e
# Calculate delay with exponential backoff and jitter
if attempt < max_retries:
delay = min(base_delay * (2 ** attempt), max_delay)
jitter = random.uniform(0, 0.1) * delay # Add up to 10% jitter
total_delay = delay + jitter
print(f"Waiting {total_delay:.1f} seconds before retry...")
time.sleep(total_delay)
# All retries exhausted
raise last_exception
# Usage
def risky_operation():
return sf.query("SELECT Id, Name FROM Account LIMIT 10")
try:
results = retry_with_backoff(risky_operation, max_retries=3)
print(f"Operation succeeded: {len(results['records'])} records")
except SalesforceError as e:
print(f"Operation failed after all retries: {e}")Context Manager for Error Handling
from contextlib import contextmanager
@contextmanager
def salesforce_error_context(operation_name="Salesforce operation"):
"""
Context manager for comprehensive Salesforce error handling.
Parameters:
- operation_name: Name of the operation for logging
"""
try:
yield
except SalesforceAuthenticationFailed as e:
print(f"{operation_name} failed: Authentication error")
print("Suggestion: Check credentials and re-authenticate")
raise
except SalesforceExpiredSession as e:
print(f"{operation_name} failed: Session expired")
print("Suggestion: Refresh session and retry")
raise
except SalesforceResourceNotFound as e:
print(f"{operation_name} failed: Resource not found")
print(f"URL: {e.url}")
print("Suggestion: Verify record IDs and object names")
raise
except SalesforceMalformedRequest as e:
print(f"{operation_name} failed: Bad request")
print(f"Status: {e.status}")
if hasattr(e, 'content'):
print(f"Details: {e.content}")
print("Suggestion: Check request format and required fields")
raise
except SalesforceRefusedRequest as e:
print(f"{operation_name} failed: Permission denied")
print("Suggestion: Check user permissions and sharing settings")
raise
except SalesforceError as e:
print(f"{operation_name} failed: General Salesforce error")
print(f"Error type: {type(e).__name__}")
print(f"Status: {e.status}")
print(f"URL: {e.url}")
raise
except Exception as e:
print(f"{operation_name} failed: Unexpected error")
print(f"Error type: {type(e).__name__}")
print(f"Message: {str(e)}")
raise
# Usage
with salesforce_error_context("Account creation"):
new_account = sf.Account.create({
'Name': 'Test Account',
'Type': 'Customer'
})
print(f"Created account: {new_account['id']}")
with salesforce_error_context("Bulk insert operation"):
results = sf.bulk.Contact.insert(contact_data)
handle_bulk_errors(results, "contact insert")Error Prevention Best Practices
Input Validation
def validate_salesforce_input(data, object_type=None):
"""
Validate data before Salesforce operations to prevent errors.
Parameters:
- data: Record data dictionary or list of records
- object_type: Salesforce object type for schema validation
Returns:
dict: Validation results with errors and warnings
"""
errors = []
warnings = []
# Handle single record or list
records = data if isinstance(data, list) else [data]
for i, record in enumerate(records):
record_prefix = f"Record {i+1}: " if len(records) > 1 else ""
# Check for None values
if not record:
errors.append(f"{record_prefix}Empty record")
continue
# Check for required Name field (common requirement)
if 'Name' in record and not record['Name']:
errors.append(f"{record_prefix}Name field is empty")
# Check field name format (no spaces, proper case)
for field_name in record.keys():
if ' ' in field_name:
warnings.append(f"{record_prefix}Field '{field_name}' contains spaces")
# Check for potential typos in common fields
common_typos = {
'name': 'Name',
'type': 'Type',
'email': 'Email',
'phone': 'Phone'
}
if field_name.lower() in common_typos and field_name != common_typos[field_name.lower()]:
warnings.append(
f"{record_prefix}Field '{field_name}' might be '{common_typos[field_name.lower()]}'"
)
# Check for very long text values
for field_name, value in record.items():
if isinstance(value, str) and len(value) > 255:
warnings.append(
f"{record_prefix}Field '{field_name}' is very long ({len(value)} chars)"
)
return {
'valid': len(errors) == 0,
'errors': errors,
'warnings': warnings,
'record_count': len(records)
}
# Usage
account_data = {
'Name': 'Test Account',
'type': 'Customer', # Should be 'Type'
'Description': 'A' * 300 # Very long description
}
validation = validate_salesforce_input(account_data)
if not validation['valid']:
print("Validation errors:")
for error in validation['errors']:
print(f" {error}")
if validation['warnings']:
print("Validation warnings:")
for warning in validation['warnings']:
print(f" {warning}")
if validation['valid']:
# Proceed with operation
try:
result = sf.Account.create(account_data)
except SalesforceError as e:
error_analysis = analyze_salesforce_error(e)Proactive Error Detection
def check_salesforce_connectivity(sf):
"""
Check Salesforce connectivity and basic functionality.
Parameters:
- sf: Salesforce client instance
Returns:
dict: Connectivity status and any issues found
"""
status = {
'connected': False,
'session_valid': False,
'api_accessible': False,
'errors': [],
'warnings': []
}
try:
# Test basic connectivity
limits = sf.limits()
status['connected'] = True
status['api_accessible'] = True
# Check API usage
if hasattr(sf, 'api_usage') and sf.api_usage:
if 'api-usage' in sf.api_usage:
usage = sf.api_usage['api-usage']
usage_percent = (usage.used / usage.total) * 100
if usage_percent > 90:
status['warnings'].append(f"API usage very high: {usage_percent:.1f}%")
elif usage_percent > 75:
status['warnings'].append(f"API usage high: {usage_percent:.1f}%")
# Test session validity
try:
sf.describe()
status['session_valid'] = True
except SalesforceExpiredSession:
status['errors'].append("Session has expired")
except SalesforceAuthenticationFailed:
status['errors'].append("Authentication failed")
except SalesforceError as e:
status['errors'].append(f"Salesforce error: {e}")
except Exception as e:
status['errors'].append(f"Connection error: {e}")
return status
# Usage
connectivity = check_salesforce_connectivity(sf)
if not connectivity['connected']:
print("Salesforce connectivity issues:")
for error in connectivity['errors']:
print(f" {error}")
else:
print("Salesforce connection OK")
if connectivity['warnings']:
print("Warnings:")
for warning in connectivity['warnings']:
print(f" {warning}")Install with Tessl CLI
npx tessl i tessl/pypi-simple-salesforce