0
# Metadata API Operations
1
2
Complete metadata management including deployment, retrieval, and CRUD operations on Salesforce metadata components like custom objects, fields, workflows, and Apex classes. The Metadata API enables programmatic management of Salesforce configuration and customizations.
3
4
## SfdcMetadataApi Class
5
6
The primary interface for Salesforce Metadata API operations, providing comprehensive metadata management capabilities.
7
8
```python { .api }
9
class SfdcMetadataApi:
10
def __init__(
11
self,
12
session,
13
session_id,
14
instance,
15
metadata_url,
16
headers,
17
api_version
18
):
19
"""
20
Initialize Metadata API interface.
21
22
Parameters:
23
- session: requests.Session object for HTTP operations
24
- session_id: Authenticated Salesforce session ID
25
- instance: Salesforce instance URL
26
- metadata_url: Metadata API endpoint URL
27
- headers: HTTP headers for authentication
28
- api_version: Salesforce API version string
29
"""
30
```
31
32
### Accessing Metadata API
33
34
The SfdcMetadataApi is accessed through the `mdapi` property of the main Salesforce client:
35
36
```python
37
from simple_salesforce import Salesforce
38
39
sf = Salesforce(username='user@example.com', password='pass', security_token='token')
40
41
# Access metadata API interface
42
mdapi = sf.mdapi
43
44
# Access specific metadata types
45
custom_objects = mdapi.CustomObject
46
apex_classes = mdapi.ApexClass
47
workflows = mdapi.Workflow
48
```
49
50
## Core Metadata Operations
51
52
Fundamental operations for metadata discovery, deployment, and retrieval.
53
54
```python { .api }
55
class SfdcMetadataApi:
56
def describe_metadata(self):
57
"""
58
Describe available metadata types and their properties.
59
60
Returns:
61
dict: Complete metadata type catalog with capabilities and properties
62
"""
63
64
def list_metadata(self, queries):
65
"""
66
List metadata components of specified types.
67
68
Parameters:
69
- queries: List of ListMetadataQuery objects or dictionaries
70
Each query contains 'type' and optional 'folder'
71
72
Returns:
73
list: Metadata component information including names and properties
74
"""
75
76
def deploy(self, zipfile, sandbox, **kwargs):
77
"""
78
Deploy metadata package to Salesforce organization.
79
80
Parameters:
81
- zipfile: ZIP file path or file-like object containing metadata
82
- sandbox: True if deploying to sandbox, False for production
83
- **kwargs: Additional deployment options (checkOnly, testLevel, etc.)
84
85
Returns:
86
dict: Deployment ID and initial status information
87
"""
88
89
def checkDeployStatus(self, asyncId, **kwargs):
90
"""
91
Check status of metadata deployment.
92
93
Parameters:
94
- asyncId: Deployment ID returned from deploy()
95
- **kwargs: Additional status check options
96
97
Returns:
98
dict: Deployment status, progress, and results
99
"""
100
101
def retrieve(self, async_process_id, **kwargs):
102
"""
103
Retrieve metadata components from Salesforce.
104
105
Parameters:
106
- async_process_id: Retrieval request ID
107
- **kwargs: Additional retrieval options
108
109
Returns:
110
dict: Retrieval status and metadata information
111
"""
112
113
def check_retrieve_status(self, async_process_id, **kwargs):
114
"""
115
Check status of metadata retrieval operation.
116
117
Parameters:
118
- async_process_id: Retrieval ID from retrieve request
119
- **kwargs: Additional status options
120
121
Returns:
122
dict: Retrieval progress and completion status
123
"""
124
125
def retrieve_zip(self, async_process_id, **kwargs):
126
"""
127
Download retrieved metadata as ZIP file.
128
129
Parameters:
130
- async_process_id: Completed retrieval ID
131
- **kwargs: Additional download options
132
133
Returns:
134
bytes: ZIP file content containing retrieved metadata
135
"""
136
137
def download_unit_test_logs(self, async_process_id):
138
"""
139
Download unit test execution logs from deployment.
140
141
Parameters:
142
- async_process_id: Deployment ID with test execution
143
144
Returns:
145
str: Unit test log content and results
146
"""
147
```
148
149
## Dynamic Metadata Type Access
150
151
The SfdcMetadataApi supports dynamic attribute access to create MetadataType instances for any metadata component type.
152
153
```python
154
# Access any metadata type
155
custom_objects = mdapi.CustomObject
156
apex_classes = mdapi.ApexClass
157
flows = mdapi.Flow
158
validation_rules = mdapi.ValidationRule
159
custom_fields = mdapi.CustomField
160
workflows = mdapi.Workflow
161
```
162
163
## MetadataType Class
164
165
Interface for CRUD operations on specific metadata component types.
166
167
```python { .api }
168
class MetadataType:
169
def __init__(self, metadata_type, mdapi):
170
"""
171
Initialize metadata type interface.
172
173
Parameters:
174
- metadata_type: Metadata type name (e.g., 'CustomObject')
175
- mdapi: Parent SfdcMetadataApi instance
176
"""
177
178
def create(self, metadata):
179
"""
180
Create new metadata components.
181
182
Parameters:
183
- metadata: List of metadata component dictionaries or single component
184
185
Returns:
186
list: Creation results with success status and any errors
187
"""
188
189
def read(self, full_names):
190
"""
191
Read existing metadata components by full name.
192
193
Parameters:
194
- full_names: List of component full names or single name string
195
196
Returns:
197
list: Retrieved metadata component definitions
198
"""
199
200
def update(self, metadata):
201
"""
202
Update existing metadata components.
203
204
Parameters:
205
- metadata: List of metadata component dictionaries or single component
206
207
Returns:
208
list: Update results with success status and any errors
209
"""
210
211
def upsert(self, metadata):
212
"""
213
Create or update metadata components (upsert operation).
214
215
Parameters:
216
- metadata: List of metadata component dictionaries or single component
217
218
Returns:
219
list: Upsert results with created/updated status and any errors
220
"""
221
222
def delete(self, full_names):
223
"""
224
Delete metadata components by full name.
225
226
Parameters:
227
- full_names: List of component full names or single name string
228
229
Returns:
230
list: Deletion results with success status and any errors
231
"""
232
233
def rename(self, old_full_name, new_full_name):
234
"""
235
Rename metadata component.
236
237
Parameters:
238
- old_full_name: Current component full name
239
- new_full_name: New component full name
240
241
Returns:
242
dict: Rename operation result
243
"""
244
245
def describe(self):
246
"""
247
Describe metadata type properties and capabilities.
248
249
Returns:
250
dict: Metadata type description including fields and relationships
251
"""
252
```
253
254
## Usage Examples
255
256
### Discovering Available Metadata
257
258
```python
259
from simple_salesforce import Salesforce
260
261
sf = Salesforce(username='user@example.com', password='pass', security_token='token')
262
mdapi = sf.mdapi
263
264
# Describe all available metadata types
265
metadata_types = mdapi.describe_metadata()
266
print(f"API supports {len(metadata_types['metadataObjects'])} metadata types")
267
268
for metadata_type in metadata_types['metadataObjects']:
269
print(f"Type: {metadata_type['xmlName']}")
270
print(f" Suffix: {metadata_type.get('suffix', 'N/A')}")
271
print(f" Directory: {metadata_type.get('directoryName', 'N/A')}")
272
273
# List specific metadata components
274
queries = [
275
{'type': 'CustomObject'},
276
{'type': 'ApexClass'},
277
{'type': 'Flow'}
278
]
279
280
components = mdapi.list_metadata(queries)
281
for component in components:
282
print(f"{component['type']}: {component['fullName']}")
283
```
284
285
### Working with Custom Objects
286
287
```python
288
# Create custom object
289
custom_object_metadata = {
290
'fullName': 'MyCustomObject__c',
291
'label': 'My Custom Object',
292
'pluralLabel': 'My Custom Objects',
293
'nameField': {
294
'type': 'Text',
295
'label': 'Name'
296
},
297
'deploymentStatus': 'Deployed',
298
'sharingModel': 'ReadWrite',
299
'description': 'Custom object created via Metadata API'
300
}
301
302
# Create the custom object
303
create_results = mdapi.CustomObject.create([custom_object_metadata])
304
if create_results[0]['success']:
305
print(f"Created custom object: {create_results[0]['fullName']}")
306
else:
307
print(f"Failed to create: {create_results[0]['errors']}")
308
309
# Read existing custom object
310
existing_objects = mdapi.CustomObject.read(['Account', 'MyCustomObject__c'])
311
for obj in existing_objects:
312
print(f"Object: {obj['fullName']}")
313
print(f"Label: {obj['label']}")
314
print(f"Fields: {len(obj.get('fields', []))}")
315
316
# Update custom object
317
updated_metadata = custom_object_metadata.copy()
318
updated_metadata['description'] = 'Updated description via Metadata API'
319
320
update_results = mdapi.CustomObject.update([updated_metadata])
321
if update_results[0]['success']:
322
print("Custom object updated successfully")
323
```
324
325
### Working with Apex Classes
326
327
```python
328
# Create Apex class
329
apex_class_metadata = {
330
'fullName': 'MyTestClass',
331
'body': '''
332
public class MyTestClass {
333
public static String getMessage() {
334
return 'Hello from Metadata API!';
335
}
336
337
@isTest
338
static void testGetMessage() {
339
String message = getMessage();
340
System.assertEquals('Hello from Metadata API!', message);
341
}
342
}
343
''',
344
'status': 'Active'
345
}
346
347
# Create the Apex class
348
apex_results = mdapi.ApexClass.create([apex_class_metadata])
349
if apex_results[0]['success']:
350
print("Apex class created successfully")
351
352
# List all Apex classes
353
apex_queries = [{'type': 'ApexClass'}]
354
apex_components = mdapi.list_metadata(apex_queries)
355
print(f"Found {len(apex_components)} Apex classes")
356
357
# Read specific Apex class
358
apex_class_def = mdapi.ApexClass.read(['MyTestClass'])
359
print(f"Apex class body:\n{apex_class_def[0]['body']}")
360
```
361
362
### Custom Fields Management
363
364
```python
365
# Create custom field
366
custom_field_metadata = {
367
'fullName': 'Account.CustomField__c',
368
'type': 'Text',
369
'label': 'Custom Field',
370
'length': 255,
371
'required': False,
372
'description': 'Custom field created via Metadata API'
373
}
374
375
# Create the field
376
field_results = mdapi.CustomField.create([custom_field_metadata])
377
if field_results[0]['success']:
378
print("Custom field created successfully")
379
380
# Update field properties
381
updated_field = custom_field_metadata.copy()
382
updated_field['description'] = 'Updated field description'
383
updated_field['required'] = True
384
385
field_update_results = mdapi.CustomField.update([updated_field])
386
print(f"Field update success: {field_update_results[0]['success']}")
387
```
388
389
### Deployment Operations
390
391
```python
392
# Deploy metadata package
393
with open('/path/to/metadata_package.zip', 'rb') as zipfile:
394
deploy_result = mdapi.deploy(
395
zipfile=zipfile,
396
sandbox=True, # Set to False for production
397
checkOnly=False, # Set to True for validation-only deployment
398
testLevel='RunLocalTests' # Test execution level
399
)
400
401
deployment_id = deploy_result['id']
402
print(f"Started deployment: {deployment_id}")
403
404
# Monitor deployment progress
405
import time
406
407
while True:
408
status = mdapi.checkDeployStatus(deployment_id)
409
410
print(f"Deployment status: {status['state']}")
411
print(f"Tests completed: {status.get('numberTestsCompleted', 0)}")
412
413
if status['done']:
414
if status['success']:
415
print("Deployment completed successfully!")
416
else:
417
print("Deployment failed:")
418
for error in status.get('details', {}).get('componentFailures', []):
419
print(f" {error['fullName']}: {error['problem']}")
420
break
421
422
time.sleep(10)
423
424
# Download test logs if available
425
if status.get('numberTestsCompleted', 0) > 0:
426
test_logs = mdapi.download_unit_test_logs(deployment_id)
427
print("Unit test logs:")
428
print(test_logs)
429
```
430
431
### Retrieval Operations
432
433
```python
434
# Create retrieval request
435
retrieval_request = {
436
'apiVersion': '59.0',
437
'types': [
438
{
439
'name': 'CustomObject',
440
'members': ['Account', 'Contact', 'MyCustomObject__c']
441
},
442
{
443
'name': 'ApexClass',
444
'members': ['*'] # Retrieve all Apex classes
445
}
446
]
447
}
448
449
# Start retrieval
450
retrieve_result = mdapi.retrieve(retrieval_request)
451
retrieval_id = retrieve_result['id']
452
453
# Wait for completion
454
while True:
455
status = mdapi.check_retrieve_status(retrieval_id)
456
457
if status['done']:
458
if status['success']:
459
print("Retrieval completed successfully")
460
461
# Download ZIP file
462
zip_content = mdapi.retrieve_zip(retrieval_id)
463
464
# Save to file
465
with open('/path/to/retrieved_metadata.zip', 'wb') as f:
466
f.write(zip_content)
467
468
print("Metadata saved to retrieved_metadata.zip")
469
else:
470
print("Retrieval failed:")
471
for message in status.get('messages', []):
472
print(f" {message['fileName']}: {message['problem']}")
473
break
474
475
time.sleep(5)
476
```
477
478
### Validation and Testing
479
480
```python
481
def validate_deployment(mdapi, zipfile_path):
482
"""Validate metadata deployment without actually deploying."""
483
484
with open(zipfile_path, 'rb') as zipfile:
485
deploy_result = mdapi.deploy(
486
zipfile=zipfile,
487
sandbox=True,
488
checkOnly=True, # Validation only
489
testLevel='RunLocalTests'
490
)
491
492
validation_id = deploy_result['id']
493
494
# Monitor validation
495
while True:
496
status = mdapi.checkDeployStatus(validation_id)
497
498
if status['done']:
499
return {
500
'valid': status['success'],
501
'errors': status.get('details', {}).get('componentFailures', []),
502
'test_results': status.get('details', {}).get('runTestResult', {}),
503
'deployment_id': validation_id
504
}
505
506
time.sleep(5)
507
508
# Usage
509
validation_result = validate_deployment(mdapi, '/path/to/package.zip')
510
if validation_result['valid']:
511
print("Package validation successful")
512
else:
513
print("Package validation failed:")
514
for error in validation_result['errors']:
515
print(f" {error['fullName']}: {error['problem']}")
516
```
517
518
### Metadata Comparison and Synchronization
519
520
```python
521
def compare_metadata_between_orgs(source_sf, target_sf, metadata_type):
522
"""Compare metadata between two Salesforce orgs."""
523
524
# List metadata in both orgs
525
source_components = source_sf.mdapi.list_metadata([{'type': metadata_type}])
526
target_components = target_sf.mdapi.list_metadata([{'type': metadata_type}])
527
528
# Create sets of component names
529
source_names = {comp['fullName'] for comp in source_components}
530
target_names = {comp['fullName'] for comp in target_components}
531
532
# Find differences
533
only_in_source = source_names - target_names
534
only_in_target = target_names - source_names
535
in_both = source_names & target_names
536
537
return {
538
'only_in_source': list(only_in_source),
539
'only_in_target': list(only_in_target),
540
'in_both': list(in_both),
541
'total_source': len(source_names),
542
'total_target': len(target_names)
543
}
544
545
# Usage
546
comparison = compare_metadata_between_orgs(source_sf, target_sf, 'ApexClass')
547
print(f"Apex classes only in source: {len(comparison['only_in_source'])}")
548
print(f"Apex classes only in target: {len(comparison['only_in_target'])}")
549
print(f"Common Apex classes: {len(comparison['in_both'])}")
550
```
551
552
### Bulk Metadata Operations
553
554
```python
555
def bulk_create_custom_fields(mdapi, object_name, field_definitions):
556
"""Create multiple custom fields for an object."""
557
558
field_metadata = []
559
560
for field_name, field_config in field_definitions.items():
561
metadata = {
562
'fullName': f'{object_name}.{field_name}',
563
'type': field_config['type'],
564
'label': field_config['label'],
565
**field_config.get('properties', {})
566
}
567
field_metadata.append(metadata)
568
569
# Create all fields in batch
570
results = mdapi.CustomField.create(field_metadata)
571
572
# Report results
573
success_count = sum(1 for r in results if r['success'])
574
575
print(f"Created {success_count}/{len(results)} custom fields")
576
577
for i, result in enumerate(results):
578
if not result['success']:
579
field_name = field_metadata[i]['fullName']
580
errors = result.get('errors', [])
581
print(f"Failed to create {field_name}: {errors}")
582
583
return results
584
585
# Usage
586
field_definitions = {
587
'CustomText__c': {
588
'type': 'Text',
589
'label': 'Custom Text Field',
590
'properties': {'length': 255, 'required': False}
591
},
592
'CustomNumber__c': {
593
'type': 'Number',
594
'label': 'Custom Number Field',
595
'properties': {'precision': 10, 'scale': 2}
596
},
597
'CustomDate__c': {
598
'type': 'Date',
599
'label': 'Custom Date Field',
600
'properties': {'required': True}
601
}
602
}
603
604
bulk_create_custom_fields(mdapi, 'Account', field_definitions)
605
```
606
607
## Advanced Metadata Operations
608
609
### Package Management
610
611
```python
612
def create_metadata_package(components, package_name, api_version='59.0'):
613
"""Create metadata package definition for deployment."""
614
615
# Group components by type
616
types_dict = {}
617
for component in components:
618
comp_type = component['type']
619
if comp_type not in types_dict:
620
types_dict[comp_type] = []
621
types_dict[comp_type].append(component['fullName'])
622
623
# Create package.xml structure
624
package_xml = {
625
'Package': {
626
'types': [
627
{
628
'name': type_name,
629
'members': members
630
}
631
for type_name, members in types_dict.items()
632
],
633
'version': api_version
634
}
635
}
636
637
return package_xml
638
639
# Usage
640
components_to_deploy = [
641
{'type': 'CustomObject', 'fullName': 'MyObject__c'},
642
{'type': 'ApexClass', 'fullName': 'MyController'},
643
{'type': 'ApexClass', 'fullName': 'MyControllerTest'}
644
]
645
646
package = create_metadata_package(components_to_deploy, 'MyDeployment')
647
```
648
649
### Error Handling and Retry Logic
650
651
```python
652
def metadata_operation_with_retry(operation_func, max_retries=3, delay=5):
653
"""Execute metadata operation with retry logic."""
654
655
for attempt in range(max_retries):
656
try:
657
result = operation_func()
658
659
# Check if operation succeeded
660
if isinstance(result, list):
661
failed_items = [r for r in result if not r['success']]
662
if failed_items:
663
print(f"Attempt {attempt + 1}: {len(failed_items)} items failed")
664
if attempt < max_retries - 1:
665
time.sleep(delay)
666
continue
667
else:
668
return result
669
else:
670
print(f"All items succeeded on attempt {attempt + 1}")
671
return result
672
else:
673
return result
674
675
except Exception as e:
676
print(f"Attempt {attempt + 1} failed with error: {e}")
677
if attempt < max_retries - 1:
678
time.sleep(delay)
679
else:
680
raise
681
682
return result
683
684
# Usage
685
def create_apex_classes():
686
return mdapi.ApexClass.create(apex_classes_metadata)
687
688
result = metadata_operation_with_retry(create_apex_classes, max_retries=3)
689
```
690
691
## Best Practices
692
693
### Metadata Deployment Safety
694
695
```python
696
def safe_metadata_deployment(mdapi, package_path, is_production=False):
697
"""Deploy metadata with proper safety checks."""
698
699
# Always validate first
700
print("Validating deployment...")
701
validation = validate_deployment(mdapi, package_path)
702
703
if not validation['valid']:
704
print("Validation failed - stopping deployment")
705
return False
706
707
# Extra safety for production
708
if is_production:
709
response = input("Deploy to PRODUCTION? Type 'DEPLOY' to confirm: ")
710
if response != 'DEPLOY':
711
print("Deployment cancelled")
712
return False
713
714
# Proceed with deployment
715
print("Starting deployment...")
716
with open(package_path, 'rb') as zipfile:
717
deploy_result = mdapi.deploy(
718
zipfile=zipfile,
719
sandbox=not is_production,
720
testLevel='RunLocalTests' if is_production else 'NoTestRun'
721
)
722
723
return monitor_deployment(mdapi, deploy_result['id'])
724
725
def monitor_deployment(mdapi, deployment_id):
726
"""Monitor deployment with detailed progress reporting."""
727
728
while True:
729
status = mdapi.checkDeployStatus(deployment_id)
730
731
# Report progress
732
total_components = status.get('numberComponentsTotal', 0)
733
deployed_components = status.get('numberComponentsDeployed', 0)
734
735
if total_components > 0:
736
progress = (deployed_components / total_components) * 100
737
print(f"Deployment progress: {progress:.1f}% ({deployed_components}/{total_components})")
738
739
if status['done']:
740
return status['success']
741
742
time.sleep(10)
743
744
# Usage
745
success = safe_metadata_deployment(mdapi, '/path/to/package.zip', is_production=False)
746
```