0
# Utility Functions
1
2
Formatting and utility functions for safe SOQL construction, external ID handling, data transformation, and error management across all Salesforce operations. These utilities ensure data safety, proper escaping, and consistent error handling throughout the simple-salesforce library.
3
4
## SOQL Formatting Functions
5
6
Safe SOQL query construction with automatic escaping and parameter binding to prevent SOQL injection attacks.
7
8
```python { .api }
9
def format_soql(query, *args, **kwargs):
10
"""
11
Format SOQL query with safe parameter substitution and escaping.
12
13
Parameters:
14
- query: SOQL query string with {} placeholders for positional args or {name} for keyword args
15
- *args: Positional arguments for substitution
16
- **kwargs: Keyword arguments for substitution
17
18
Returns:
19
str: Safely formatted SOQL query with properly escaped values
20
21
Examples:
22
format_soql("SELECT Id FROM Account WHERE Name = {}", "O'Reilly Corp")
23
format_soql("SELECT Id FROM Account WHERE Name = {name} AND Type = {type}",
24
name="Test Corp", type="Customer")
25
"""
26
27
def quote_soql_value(value):
28
"""
29
Quote and escape individual value for safe SOQL usage.
30
31
Parameters:
32
- value: Value to quote and escape for SOQL
33
34
Returns:
35
str: Properly quoted and escaped value safe for SOQL queries
36
37
Examples:
38
quote_soql_value("O'Reilly") -> "'O\\'Reilly'"
39
quote_soql_value("Test Corp") -> "'Test Corp'"
40
"""
41
42
def format_external_id(field, value):
43
"""
44
Create external ID string for upsert operations.
45
46
Parameters:
47
- field: External ID field API name
48
- value: External ID value
49
50
Returns:
51
str: Formatted external ID string for upsert operations
52
53
Examples:
54
format_external_id("External_ID__c", "EXT123") -> "External_ID__c/EXT123"
55
"""
56
```
57
58
## SoqlFormatter Class
59
60
Advanced SOQL formatting with custom format specifications for different value types.
61
62
```python { .api }
63
class SoqlFormatter:
64
def __init__(self):
65
"""
66
Custom formatter for SOQL queries with special format specifications.
67
68
Supports:
69
- :literal - Insert value without quoting (for field names, functions)
70
- :like - Escape value for LIKE expressions with % and _ wildcards
71
"""
72
73
def format(self, format_string, *args, **kwargs):
74
"""
75
Format string with SOQL-specific format specifications.
76
77
Parameters:
78
- format_string: Format string with placeholders and format specs
79
- *args: Positional arguments
80
- **kwargs: Keyword arguments
81
82
Returns:
83
str: Formatted string with SOQL-safe values
84
"""
85
```
86
87
## Usage Examples
88
89
### Basic SOQL Formatting
90
91
```python
92
from simple_salesforce import format_soql, quote_soql_value
93
94
# Simple parameter substitution
95
account_name = "O'Reilly Corporation"
96
query = format_soql("SELECT Id, Name FROM Account WHERE Name = {}", account_name)
97
print(query)
98
# Output: SELECT Id, Name FROM Account WHERE Name = 'O\'Reilly Corporation'
99
100
# Multiple parameters with keywords
101
query = format_soql(
102
"SELECT Id, Name FROM Account WHERE Name = {name} AND Type = {account_type}",
103
name="Test Corp",
104
account_type="Customer"
105
)
106
print(query)
107
# Output: SELECT Id, Name FROM Account WHERE Name = 'Test Corp' AND Type = 'Customer'
108
109
# Numeric and boolean values
110
query = format_soql(
111
"SELECT Id FROM Account WHERE NumberOfEmployees > {} AND IsPersonAccount = {}",
112
100,
113
False
114
)
115
print(query)
116
# Output: SELECT Id FROM Account WHERE NumberOfEmployees > 100 AND IsPersonAccount = false
117
118
# Date and datetime handling
119
from datetime import datetime, date
120
121
query = format_soql(
122
"SELECT Id FROM Account WHERE CreatedDate >= {} AND LastModifiedDate = {}",
123
datetime(2023, 1, 1),
124
date.today()
125
)
126
print(query)
127
# Output includes properly formatted ISO dates
128
```
129
130
### Advanced SOQL Formatting with Format Specs
131
132
```python
133
from simple_salesforce.format import SoqlFormatter
134
135
formatter = SoqlFormatter()
136
137
# Literal values (field names, functions) - no quoting
138
field_name = "Custom_Field__c"
139
query = formatter.format(
140
"SELECT Id, {field:literal} FROM Account ORDER BY {field:literal}",
141
field=field_name
142
)
143
print(query)
144
# Output: SELECT Id, Custom_Field__c FROM Account ORDER BY Custom_Field__c
145
146
# LIKE expressions with proper escaping
147
search_term = "Test% Corp_Name"
148
query = formatter.format(
149
"SELECT Id FROM Account WHERE Name LIKE '{search:like}'",
150
search=search_term
151
)
152
print(query)
153
# Output: SELECT Id FROM Account WHERE Name LIKE 'Test\\% Corp\\_Name'
154
155
# Combined usage
156
query = formatter.format(
157
"SELECT {fields:literal} FROM {object:literal} WHERE Name LIKE '{pattern:like}' AND Type = '{type}'",
158
fields="Id, Name, Type",
159
object="Account",
160
pattern="%Corp%",
161
type="Customer"
162
)
163
```
164
165
### External ID Formatting
166
167
```python
168
from simple_salesforce import format_external_id
169
170
# Create external ID references for upsert
171
external_ref = format_external_id("External_ID__c", "EXT123")
172
print(external_ref)
173
# Output: External_ID__c/EXT123
174
175
# Usage in upsert operations
176
sf = Salesforce(username='user@example.com', password='pass', security_token='token')
177
178
# Upsert using external ID
179
result = sf.Account.upsert(
180
external_ref, # Uses External_ID__c/EXT123
181
{
182
'Name': 'Updated Account Name',
183
'Type': 'Customer'
184
}
185
)
186
187
# Bulk upsert with external IDs
188
upsert_data = []
189
for i, external_id in enumerate(['EXT001', 'EXT002', 'EXT003']):
190
upsert_data.append({
191
'External_ID__c': external_id,
192
'Name': f'Account {i+1}',
193
'Type': 'Customer'
194
})
195
196
bulk_results = sf.bulk.Account.upsert(
197
upsert_data,
198
external_id_field='External_ID__c'
199
)
200
```
201
202
### Safe Query Construction Patterns
203
204
```python
205
def build_dynamic_query(base_fields, conditions, order_by=None):
206
"""Build SOQL query dynamically with safe parameter handling."""
207
208
# Start with base query
209
query_parts = ["SELECT"]
210
211
# Add fields safely
212
if isinstance(base_fields, list):
213
fields = ', '.join(base_fields)
214
else:
215
fields = base_fields
216
217
query_parts.append(fields)
218
query_parts.append("FROM Account")
219
220
# Add WHERE conditions safely
221
if conditions:
222
where_clauses = []
223
values = []
224
225
for condition in conditions:
226
if condition['operator'] == '=':
227
where_clauses.append(f"{condition['field']} = {{}}")
228
values.append(condition['value'])
229
elif condition['operator'] == 'LIKE':
230
where_clauses.append(f"{condition['field']} LIKE {{}}")
231
values.append(f"%{condition['value']}%")
232
elif condition['operator'] == 'IN':
233
placeholders = ', '.join(['{}'] * len(condition['value']))
234
where_clauses.append(f"{condition['field']} IN ({placeholders})")
235
values.extend(condition['value'])
236
237
if where_clauses:
238
query_parts.append("WHERE")
239
query_parts.append(" AND ".join(where_clauses))
240
241
# Add ORDER BY
242
if order_by:
243
query_parts.append(f"ORDER BY {order_by}")
244
245
# Format safely
246
base_query = " ".join(query_parts)
247
return format_soql(base_query, *values)
248
249
# Usage
250
conditions = [
251
{'field': 'Name', 'operator': 'LIKE', 'value': 'Corp'},
252
{'field': 'Type', 'operator': '=', 'value': 'Customer'},
253
{'field': 'Industry', 'operator': 'IN', 'value': ['Technology', 'Healthcare']}
254
]
255
256
safe_query = build_dynamic_query(
257
base_fields=['Id', 'Name', 'Type', 'Industry'],
258
conditions=conditions,
259
order_by='Name'
260
)
261
262
# Execute safely
263
results = sf.query(safe_query)
264
```
265
266
## Type Definitions and Constants
267
268
Utility types and constants used throughout the simple-salesforce library.
269
270
```python { .api }
271
# Type aliases for better code documentation
272
Headers = dict # Type alias for MutableMapping[str, str]
273
Proxies = dict # Type alias for MutableMapping[str, str]
274
BulkDataAny = list # Type alias for List[Mapping[str, Any]]
275
BulkDataStr = list # Type alias for List[Mapping[str, str]]
276
277
# Named tuples for structured data
278
class Usage:
279
"""
280
API usage information from Salesforce headers.
281
282
Attributes:
283
- used: Number of API calls used
284
- total: Total API calls allowed
285
"""
286
def __init__(self, used: int, total: int):
287
self.used = used
288
self.total = total
289
290
class PerAppUsage:
291
"""
292
Per-application API usage information.
293
294
Attributes:
295
- used: API calls used by this app
296
- total: Total API calls allowed for this app
297
- name: Application name
298
"""
299
def __init__(self, used: int, total: int, name: str):
300
self.used = used
301
self.total = total
302
self.name = name
303
304
# Escaping constants
305
soql_escapes = str.maketrans({
306
"'": "\\'",
307
"\\": "\\\\",
308
"\n": "\\n",
309
"\r": "\\r",
310
"\t": "\\t",
311
"\b": "\\b",
312
"\f": "\\f",
313
'"': '\\"'
314
})
315
316
soql_like_escapes = str.maketrans({
317
"%": "\\%",
318
"_": "\\_",
319
"'": "\\'",
320
"\\": "\\\\"
321
})
322
```
323
324
## Utility Functions
325
326
Helper functions for XML parsing, date handling, HTTP operations, and error management.
327
328
```python { .api }
329
def getUniqueElementValueFromXmlString(xmlString, elementName):
330
"""
331
Extract single element value from XML string.
332
333
Parameters:
334
- xmlString: XML content as string
335
- elementName: Element tag name to extract
336
337
Returns:
338
str: Element text content or None if not found
339
340
Example:
341
xml = "<response><sessionId>abc123</sessionId></response>"
342
session_id = getUniqueElementValueFromXmlString(xml, "sessionId")
343
"""
344
345
def date_to_iso8601(date):
346
"""
347
Convert date object to ISO8601 string with URL encoding.
348
349
Parameters:
350
- date: datetime.date or datetime.datetime object
351
352
Returns:
353
str: URL-encoded ISO8601 date string suitable for Salesforce APIs
354
355
Example:
356
from datetime import datetime
357
iso_date = date_to_iso8601(datetime(2023, 12, 25, 14, 30))
358
# Returns: "2023-12-25T14%3A30%3A00"
359
"""
360
361
def exception_handler(result, name=""):
362
"""
363
Route HTTP errors to appropriate Salesforce exception types.
364
365
Parameters:
366
- result: requests.Response object
367
- name: Optional operation name for error context
368
369
Raises:
370
- SalesforceError: Base exception for Salesforce API errors
371
- Specific exceptions based on HTTP status codes
372
373
Example:
374
response = requests.get(url)
375
exception_handler(response, "Account query")
376
"""
377
378
def call_salesforce(url, method, session, headers, **kwargs):
379
"""
380
Utility function for making HTTP calls to Salesforce with proper error handling.
381
382
Parameters:
383
- url: Salesforce API endpoint URL
384
- method: HTTP method (GET, POST, PUT, PATCH, DELETE)
385
- session: requests.Session object
386
- headers: HTTP headers dictionary
387
- **kwargs: Additional request parameters (data, json, params, etc.)
388
389
Returns:
390
requests.Response: HTTP response object
391
392
Raises:
393
- Appropriate Salesforce exceptions for error responses
394
"""
395
396
def list_from_generator(generator_function):
397
"""
398
Convert generator function result to list.
399
400
Parameters:
401
- generator_function: Function that returns a generator
402
403
Returns:
404
list: All items from the generator as a list
405
406
Example:
407
def query_generator():
408
for i in range(1000):
409
yield f"Record {i}"
410
411
all_records = list_from_generator(query_generator)
412
"""
413
```
414
415
## Advanced Usage Examples
416
417
### Custom SOQL Builder Class
418
419
```python
420
class SafeSOQLBuilder:
421
"""Builder for constructing SOQL queries safely."""
422
423
def __init__(self, sobject):
424
self.sobject = sobject
425
self.fields = []
426
self.conditions = []
427
self.order_fields = []
428
self.limit_value = None
429
430
def select(self, *fields):
431
"""Add fields to SELECT clause."""
432
self.fields.extend(fields)
433
return self
434
435
def where(self, field, operator, value):
436
"""Add WHERE condition."""
437
self.conditions.append((field, operator, value))
438
return self
439
440
def order_by(self, field, direction='ASC'):
441
"""Add ORDER BY field."""
442
self.order_fields.append(f"{field} {direction}")
443
return self
444
445
def limit(self, count):
446
"""Set LIMIT clause."""
447
self.limit_value = count
448
return self
449
450
def build(self):
451
"""Build the final SOQL query."""
452
if not self.fields:
453
raise ValueError("No fields specified for SELECT")
454
455
# Build query parts
456
query_parts = []
457
values = []
458
459
# SELECT clause
460
fields_str = ', '.join(self.fields)
461
query_parts.append(f"SELECT {fields_str}")
462
463
# FROM clause
464
query_parts.append(f"FROM {self.sobject}")
465
466
# WHERE clause
467
if self.conditions:
468
where_parts = []
469
for field, operator, value in self.conditions:
470
if operator.upper() == 'IN' and isinstance(value, (list, tuple)):
471
placeholders = ', '.join(['{}'] * len(value))
472
where_parts.append(f"{field} IN ({placeholders})")
473
values.extend(value)
474
else:
475
where_parts.append(f"{field} {operator} {{}}")
476
values.append(value)
477
478
query_parts.append(f"WHERE {' AND '.join(where_parts)}")
479
480
# ORDER BY clause
481
if self.order_fields:
482
query_parts.append(f"ORDER BY {', '.join(self.order_fields)}")
483
484
# LIMIT clause
485
if self.limit_value:
486
query_parts.append(f"LIMIT {self.limit_value}")
487
488
# Format safely
489
base_query = ' '.join(query_parts)
490
return format_soql(base_query, *values)
491
492
# Usage
493
query = (SafeSOQLBuilder('Account')
494
.select('Id', 'Name', 'Type', 'Industry')
495
.where('Name', 'LIKE', '%Corp%')
496
.where('Type', '=', 'Customer')
497
.where('Industry', 'IN', ['Technology', 'Healthcare'])
498
.order_by('Name')
499
.limit(100)
500
.build())
501
502
results = sf.query(query)
503
```
504
505
### Bulk Data Preparation Utilities
506
507
```python
508
def prepare_bulk_data(records, field_mapping=None, default_values=None):
509
"""
510
Prepare record data for bulk operations with validation and transformation.
511
512
Parameters:
513
- records: List of record dictionaries
514
- field_mapping: Dictionary mapping source fields to Salesforce fields
515
- default_values: Dictionary of default values for missing fields
516
517
Returns:
518
list: Cleaned and validated records ready for bulk operations
519
"""
520
521
prepared_records = []
522
field_mapping = field_mapping or {}
523
default_values = default_values or {}
524
525
for record in records:
526
# Apply field mapping
527
mapped_record = {}
528
for source_field, value in record.items():
529
target_field = field_mapping.get(source_field, source_field)
530
mapped_record[target_field] = value
531
532
# Apply default values
533
for field, default_value in default_values.items():
534
if field not in mapped_record or mapped_record[field] is None:
535
mapped_record[field] = default_value
536
537
# Clean null values
538
cleaned_record = {
539
k: v for k, v in mapped_record.items()
540
if v is not None and v != ''
541
}
542
543
# Validate required fields (example)
544
if 'Name' not in cleaned_record:
545
cleaned_record['Name'] = 'Unnamed Record'
546
547
prepared_records.append(cleaned_record)
548
549
return prepared_records
550
551
def validate_record_data(records, required_fields=None, field_types=None):
552
"""
553
Validate record data before bulk operations.
554
555
Parameters:
556
- records: List of record dictionaries
557
- required_fields: List of required field names
558
- field_types: Dictionary mapping field names to expected types
559
560
Returns:
561
dict: Validation results with errors and warnings
562
"""
563
564
required_fields = required_fields or []
565
field_types = field_types or {}
566
567
errors = []
568
warnings = []
569
570
for i, record in enumerate(records):
571
# Check required fields
572
for field in required_fields:
573
if field not in record or not record[field]:
574
errors.append(f"Record {i+1}: Missing required field '{field}'")
575
576
# Check field types
577
for field, expected_type in field_types.items():
578
if field in record and record[field] is not None:
579
if not isinstance(record[field], expected_type):
580
warnings.append(
581
f"Record {i+1}: Field '{field}' expected {expected_type.__name__}, "
582
f"got {type(record[field]).__name__}"
583
)
584
585
return {
586
'valid': len(errors) == 0,
587
'errors': errors,
588
'warnings': warnings,
589
'total_records': len(records)
590
}
591
592
# Usage
593
raw_data = [
594
{'company_name': 'Test Corp', 'account_type': 'Customer'},
595
{'company_name': 'Another Corp', 'account_type': 'Partner'},
596
# ... more records
597
]
598
599
prepared_data = prepare_bulk_data(
600
raw_data,
601
field_mapping={'company_name': 'Name', 'account_type': 'Type'},
602
default_values={'Industry': 'Other'}
603
)
604
605
validation = validate_record_data(
606
prepared_data,
607
required_fields=['Name'],
608
field_types={'Name': str, 'Type': str}
609
)
610
611
if validation['valid']:
612
results = sf.bulk.Account.insert(prepared_data)
613
else:
614
print("Validation errors:")
615
for error in validation['errors']:
616
print(f" {error}")
617
```
618
619
### Error Recovery Utilities
620
621
```python
622
def retry_failed_records(bulk_operation_func, records, max_retries=3):
623
"""
624
Retry failed records from bulk operations with exponential backoff.
625
626
Parameters:
627
- bulk_operation_func: Function that performs the bulk operation
628
- records: Initial record data
629
- max_retries: Maximum number of retry attempts
630
631
Returns:
632
dict: Final results with success/failure counts
633
"""
634
635
current_records = records.copy()
636
all_results = []
637
retry_count = 0
638
639
while current_records and retry_count < max_retries:
640
print(f"Attempt {retry_count + 1}: Processing {len(current_records)} records")
641
642
try:
643
# Perform bulk operation
644
results = bulk_operation_func(current_records)
645
all_results.extend(results)
646
647
# Identify failed records for retry
648
failed_records = []
649
for i, result in enumerate(results):
650
if not result.get('success', False):
651
failed_records.append(current_records[i])
652
print(f"Failed record: {result.get('error', 'Unknown error')}")
653
654
current_records = failed_records
655
retry_count += 1
656
657
if failed_records:
658
# Exponential backoff
659
wait_time = 2 ** retry_count
660
print(f"Waiting {wait_time} seconds before retry...")
661
time.sleep(wait_time)
662
663
except Exception as e:
664
print(f"Bulk operation failed: {e}")
665
retry_count += 1
666
if retry_count < max_retries:
667
wait_time = 2 ** retry_count
668
time.sleep(wait_time)
669
else:
670
raise
671
672
# Calculate final statistics
673
successful = sum(1 for r in all_results if r.get('success', False))
674
failed = len(all_results) - successful
675
676
return {
677
'total_processed': len(records),
678
'successful': successful,
679
'failed': failed,
680
'success_rate': (successful / len(records)) * 100 if records else 0,
681
'retry_attempts': retry_count,
682
'final_results': all_results
683
}
684
685
# Usage
686
def perform_bulk_insert(records):
687
return sf.bulk.Account.insert(records, include_detailed_results=True)
688
689
final_results = retry_failed_records(
690
perform_bulk_insert,
691
prepared_account_data,
692
max_retries=3
693
)
694
695
print(f"Final success rate: {final_results['success_rate']:.1f}%")
696
```
697
698
## Performance and Monitoring Utilities
699
700
```python
701
def monitor_api_usage(sf, operation_name="Operation"):
702
"""
703
Monitor and report API usage after operations.
704
705
Parameters:
706
- sf: Salesforce client instance
707
- operation_name: Name of operation for logging
708
"""
709
710
if hasattr(sf, 'api_usage') and sf.api_usage:
711
usage_info = sf.api_usage
712
713
print(f"{operation_name} API Usage:")
714
if 'api-usage' in usage_info:
715
total_usage = usage_info['api-usage']
716
print(f" Total API calls: {total_usage.used}/{total_usage.total}")
717
718
if 'per-app-api-usage' in usage_info:
719
for app_usage in usage_info['per-app-api-usage']:
720
print(f" {app_usage.name}: {app_usage.used}/{app_usage.total}")
721
722
def benchmark_query_performance(sf, queries, iterations=1):
723
"""
724
Benchmark query performance for optimization.
725
726
Parameters:
727
- sf: Salesforce client instance
728
- queries: List of SOQL queries to benchmark
729
- iterations: Number of times to run each query
730
731
Returns:
732
list: Performance results for each query
733
"""
734
735
results = []
736
737
for query in queries:
738
times = []
739
740
for i in range(iterations):
741
start_time = time.time()
742
743
try:
744
result = sf.query(query)
745
end_time = time.time()
746
747
execution_time = end_time - start_time
748
times.append(execution_time)
749
750
print(f"Query {len(results)+1}, Run {i+1}: {execution_time:.3f}s "
751
f"({result['totalSize']} records)")
752
753
except Exception as e:
754
print(f"Query {len(results)+1}, Run {i+1}: Failed - {e}")
755
times.append(None)
756
757
# Calculate statistics
758
valid_times = [t for t in times if t is not None]
759
if valid_times:
760
avg_time = sum(valid_times) / len(valid_times)
761
min_time = min(valid_times)
762
max_time = max(valid_times)
763
else:
764
avg_time = min_time = max_time = None
765
766
results.append({
767
'query': query,
768
'iterations': iterations,
769
'avg_time': avg_time,
770
'min_time': min_time,
771
'max_time': max_time,
772
'success_rate': len(valid_times) / iterations * 100
773
})
774
775
return results
776
777
# Usage
778
queries_to_test = [
779
"SELECT Id, Name FROM Account LIMIT 100",
780
"SELECT Id, Name FROM Account WHERE CreatedDate = TODAY",
781
"SELECT Id, Name, (SELECT Id, Name FROM Contacts) FROM Account LIMIT 50"
782
]
783
784
performance_results = benchmark_query_performance(sf, queries_to_test, iterations=3)
785
786
for result in performance_results:
787
print(f"Query: {result['query'][:50]}...")
788
print(f" Average time: {result['avg_time']:.3f}s")
789
print(f" Success rate: {result['success_rate']:.1f}%")
790
```