Metadata API Operations

class SfdcMetadataApi:
    def __init__(
        self,
        session,
        session_id,
        instance,
        metadata_url,
        headers,
        api_version
    ):
        """
        Initialize Metadata API interface.

        Parameters:
        - session: requests.Session object for HTTP operations
        - session_id: Authenticated Salesforce session ID
        - instance: Salesforce instance URL
        - metadata_url: Metadata API endpoint URL
        - headers: HTTP headers for authentication
        - api_version: Salesforce API version string
        """

from simple_salesforce import Salesforce

sf = Salesforce(username='user@example.com', password='pass', security_token='token')

# Access metadata API interface
mdapi = sf.mdapi

# Access specific metadata types
custom_objects = mdapi.CustomObject
apex_classes = mdapi.ApexClass
workflows = mdapi.Workflow

class SfdcMetadataApi:
    def describe_metadata(self):
        """
        Describe available metadata types and their properties.

        Returns:
        dict: Complete metadata type catalog with capabilities and properties
        """

    def list_metadata(self, queries):
        """
        List metadata components of specified types.

        Parameters:
        - queries: List of ListMetadataQuery objects or dictionaries
                  Each query contains 'type' and optional 'folder'

        Returns:
        list: Metadata component information including names and properties
        """

    def deploy(self, zipfile, sandbox, **kwargs):
        """
        Deploy metadata package to Salesforce organization.

        Parameters:
        - zipfile: ZIP file path or file-like object containing metadata
        - sandbox: True if deploying to sandbox, False for production
        - **kwargs: Additional deployment options (checkOnly, testLevel, etc.)

        Returns:
        dict: Deployment ID and initial status information
        """

    def checkDeployStatus(self, asyncId, **kwargs):
        """
        Check status of metadata deployment.

        Parameters:
        - asyncId: Deployment ID returned from deploy()
        - **kwargs: Additional status check options

        Returns:
        dict: Deployment status, progress, and results
        """

    def retrieve(self, async_process_id, **kwargs):
        """
        Retrieve metadata components from Salesforce.

        Parameters:
        - async_process_id: Retrieval request ID
        - **kwargs: Additional retrieval options

        Returns:
        dict: Retrieval status and metadata information
        """

    def check_retrieve_status(self, async_process_id, **kwargs):
        """
        Check status of metadata retrieval operation.

        Parameters:
        - async_process_id: Retrieval ID from retrieve request
        - **kwargs: Additional status options

        Returns:
        dict: Retrieval progress and completion status
        """

    def retrieve_zip(self, async_process_id, **kwargs):
        """
        Download retrieved metadata as ZIP file.

        Parameters:
        - async_process_id: Completed retrieval ID
        - **kwargs: Additional download options

        Returns:
        bytes: ZIP file content containing retrieved metadata
        """

    def download_unit_test_logs(self, async_process_id):
        """
        Download unit test execution logs from deployment.

        Parameters:
        - async_process_id: Deployment ID with test execution

        Returns:
        str: Unit test log content and results
        """

# Access any metadata type
custom_objects = mdapi.CustomObject
apex_classes = mdapi.ApexClass
flows = mdapi.Flow
validation_rules = mdapi.ValidationRule
custom_fields = mdapi.CustomField
workflows = mdapi.Workflow

class MetadataType:
    def __init__(self, metadata_type, mdapi):
        """
        Initialize metadata type interface.

        Parameters:
        - metadata_type: Metadata type name (e.g., 'CustomObject')
        - mdapi: Parent SfdcMetadataApi instance
        """
        
    def create(self, metadata):
        """
        Create new metadata components.

        Parameters:
        - metadata: List of metadata component dictionaries or single component

        Returns:
        list: Creation results with success status and any errors
        """

    def read(self, full_names):
        """
        Read existing metadata components by full name.

        Parameters:
        - full_names: List of component full names or single name string

        Returns:
        list: Retrieved metadata component definitions
        """

    def update(self, metadata):
        """
        Update existing metadata components.

        Parameters:
        - metadata: List of metadata component dictionaries or single component

        Returns:
        list: Update results with success status and any errors
        """

    def upsert(self, metadata):
        """
        Create or update metadata components (upsert operation).

        Parameters:
        - metadata: List of metadata component dictionaries or single component

        Returns:
        list: Upsert results with created/updated status and any errors
        """

    def delete(self, full_names):
        """
        Delete metadata components by full name.

        Parameters:
        - full_names: List of component full names or single name string

        Returns:
        list: Deletion results with success status and any errors
        """

    def rename(self, old_full_name, new_full_name):
        """
        Rename metadata component.

        Parameters:
        - old_full_name: Current component full name
        - new_full_name: New component full name

        Returns:
        dict: Rename operation result
        """

    def describe(self):
        """
        Describe metadata type properties and capabilities.

        Returns:
        dict: Metadata type description including fields and relationships
        """

from simple_salesforce import Salesforce

sf = Salesforce(username='user@example.com', password='pass', security_token='token')
mdapi = sf.mdapi

# Describe all available metadata types
metadata_types = mdapi.describe_metadata()
print(f"API supports {len(metadata_types['metadataObjects'])} metadata types")

for metadata_type in metadata_types['metadataObjects']:
    print(f"Type: {metadata_type['xmlName']}")
    print(f"  Suffix: {metadata_type.get('suffix', 'N/A')}")
    print(f"  Directory: {metadata_type.get('directoryName', 'N/A')}")

# List specific metadata components
queries = [
    {'type': 'CustomObject'},
    {'type': 'ApexClass'},
    {'type': 'Flow'}
]

components = mdapi.list_metadata(queries)
for component in components:
    print(f"{component['type']}: {component['fullName']}")

# Create custom object
custom_object_metadata = {
    'fullName': 'MyCustomObject__c',
    'label': 'My Custom Object',
    'pluralLabel': 'My Custom Objects',
    'nameField': {
        'type': 'Text',
        'label': 'Name'
    },
    'deploymentStatus': 'Deployed',
    'sharingModel': 'ReadWrite',
    'description': 'Custom object created via Metadata API'
}

# Create the custom object
create_results = mdapi.CustomObject.create([custom_object_metadata])
if create_results[0]['success']:
    print(f"Created custom object: {create_results[0]['fullName']}")
else:
    print(f"Failed to create: {create_results[0]['errors']}")

# Read existing custom object
existing_objects = mdapi.CustomObject.read(['Account', 'MyCustomObject__c'])
for obj in existing_objects:
    print(f"Object: {obj['fullName']}")
    print(f"Label: {obj['label']}")
    print(f"Fields: {len(obj.get('fields', []))}")

# Update custom object
updated_metadata = custom_object_metadata.copy()
updated_metadata['description'] = 'Updated description via Metadata API'

update_results = mdapi.CustomObject.update([updated_metadata])
if update_results[0]['success']:
    print("Custom object updated successfully")

# Create Apex class
apex_class_metadata = {
    'fullName': 'MyTestClass',
    'body': '''
public class MyTestClass {
    public static String getMessage() {
        return 'Hello from Metadata API!';
    }
    
    @isTest
    static void testGetMessage() {
        String message = getMessage();
        System.assertEquals('Hello from Metadata API!', message);
    }
}
''',
    'status': 'Active'
}

# Create the Apex class
apex_results = mdapi.ApexClass.create([apex_class_metadata])
if apex_results[0]['success']:
    print("Apex class created successfully")

# List all Apex classes
apex_queries = [{'type': 'ApexClass'}]
apex_components = mdapi.list_metadata(apex_queries)
print(f"Found {len(apex_components)} Apex classes")

# Read specific Apex class
apex_class_def = mdapi.ApexClass.read(['MyTestClass'])
print(f"Apex class body:\n{apex_class_def[0]['body']}")

# Create custom field
custom_field_metadata = {
    'fullName': 'Account.CustomField__c',
    'type': 'Text',
    'label': 'Custom Field',
    'length': 255,
    'required': False,
    'description': 'Custom field created via Metadata API'
}

# Create the field
field_results = mdapi.CustomField.create([custom_field_metadata])
if field_results[0]['success']:
    print("Custom field created successfully")

# Update field properties
updated_field = custom_field_metadata.copy()
updated_field['description'] = 'Updated field description'
updated_field['required'] = True

field_update_results = mdapi.CustomField.update([updated_field])
print(f"Field update success: {field_update_results[0]['success']}")

# Deploy metadata package
with open('/path/to/metadata_package.zip', 'rb') as zipfile:
    deploy_result = mdapi.deploy(
        zipfile=zipfile,
        sandbox=True,  # Set to False for production
        checkOnly=False,  # Set to True for validation-only deployment
        testLevel='RunLocalTests'  # Test execution level
    )

deployment_id = deploy_result['id']
print(f"Started deployment: {deployment_id}")

# Monitor deployment progress
import time

while True:
    status = mdapi.checkDeployStatus(deployment_id)
    
    print(f"Deployment status: {status['state']}")
    print(f"Tests completed: {status.get('numberTestsCompleted', 0)}")
    
    if status['done']:
        if status['success']:
            print("Deployment completed successfully!")
        else:
            print("Deployment failed:")
            for error in status.get('details', {}).get('componentFailures', []):
                print(f"  {error['fullName']}: {error['problem']}")
        break
    
    time.sleep(10)

# Download test logs if available
if status.get('numberTestsCompleted', 0) > 0:
    test_logs = mdapi.download_unit_test_logs(deployment_id)
    print("Unit test logs:")
    print(test_logs)

# Create retrieval request
retrieval_request = {
    'apiVersion': '59.0',
    'types': [
        {
            'name': 'CustomObject',
            'members': ['Account', 'Contact', 'MyCustomObject__c']
        },
        {
            'name': 'ApexClass', 
            'members': ['*']  # Retrieve all Apex classes
        }
    ]
}

# Start retrieval
retrieve_result = mdapi.retrieve(retrieval_request)
retrieval_id = retrieve_result['id']

# Wait for completion
while True:
    status = mdapi.check_retrieve_status(retrieval_id)
    
    if status['done']:
        if status['success']:
            print("Retrieval completed successfully")
            
            # Download ZIP file
            zip_content = mdapi.retrieve_zip(retrieval_id)
            
            # Save to file
            with open('/path/to/retrieved_metadata.zip', 'wb') as f:
                f.write(zip_content)
            
            print("Metadata saved to retrieved_metadata.zip")
        else:
            print("Retrieval failed:")
            for message in status.get('messages', []):
                print(f"  {message['fileName']}: {message['problem']}")
        break
    
    time.sleep(5)

def validate_deployment(mdapi, zipfile_path):
    """Validate metadata deployment without actually deploying."""
    
    with open(zipfile_path, 'rb') as zipfile:
        deploy_result = mdapi.deploy(
            zipfile=zipfile,
            sandbox=True,
            checkOnly=True,  # Validation only
            testLevel='RunLocalTests'
        )
    
    validation_id = deploy_result['id']
    
    # Monitor validation
    while True:
        status = mdapi.checkDeployStatus(validation_id)
        
        if status['done']:
            return {
                'valid': status['success'],
                'errors': status.get('details', {}).get('componentFailures', []),
                'test_results': status.get('details', {}).get('runTestResult', {}),
                'deployment_id': validation_id
            }
        
        time.sleep(5)

# Usage
validation_result = validate_deployment(mdapi, '/path/to/package.zip')
if validation_result['valid']:
    print("Package validation successful")
else:
    print("Package validation failed:")
    for error in validation_result['errors']:
        print(f"  {error['fullName']}: {error['problem']}")

def compare_metadata_between_orgs(source_sf, target_sf, metadata_type):
    """Compare metadata between two Salesforce orgs."""
    
    # List metadata in both orgs
    source_components = source_sf.mdapi.list_metadata([{'type': metadata_type}])
    target_components = target_sf.mdapi.list_metadata([{'type': metadata_type}])
    
    # Create sets of component names
    source_names = {comp['fullName'] for comp in source_components}
    target_names = {comp['fullName'] for comp in target_components}
    
    # Find differences
    only_in_source = source_names - target_names
    only_in_target = target_names - source_names
    in_both = source_names & target_names
    
    return {
        'only_in_source': list(only_in_source),
        'only_in_target': list(only_in_target),
        'in_both': list(in_both),
        'total_source': len(source_names),
        'total_target': len(target_names)
    }

# Usage
comparison = compare_metadata_between_orgs(source_sf, target_sf, 'ApexClass')
print(f"Apex classes only in source: {len(comparison['only_in_source'])}")
print(f"Apex classes only in target: {len(comparison['only_in_target'])}")
print(f"Common Apex classes: {len(comparison['in_both'])}")

def bulk_create_custom_fields(mdapi, object_name, field_definitions):
    """Create multiple custom fields for an object."""
    
    field_metadata = []
    
    for field_name, field_config in field_definitions.items():
        metadata = {
            'fullName': f'{object_name}.{field_name}',
            'type': field_config['type'],
            'label': field_config['label'],
            **field_config.get('properties', {})
        }
        field_metadata.append(metadata)
    
    # Create all fields in batch
    results = mdapi.CustomField.create(field_metadata)
    
    # Report results
    success_count = sum(1 for r in results if r['success'])
    
    print(f"Created {success_count}/{len(results)} custom fields")
    
    for i, result in enumerate(results):
        if not result['success']:
            field_name = field_metadata[i]['fullName']
            errors = result.get('errors', [])
            print(f"Failed to create {field_name}: {errors}")
    
    return results

# Usage
field_definitions = {
    'CustomText__c': {
        'type': 'Text',
        'label': 'Custom Text Field',
        'properties': {'length': 255, 'required': False}
    },
    'CustomNumber__c': {
        'type': 'Number',
        'label': 'Custom Number Field', 
        'properties': {'precision': 10, 'scale': 2}
    },
    'CustomDate__c': {
        'type': 'Date',
        'label': 'Custom Date Field',
        'properties': {'required': True}
    }
}

bulk_create_custom_fields(mdapi, 'Account', field_definitions)

def create_metadata_package(components, package_name, api_version='59.0'):
    """Create metadata package definition for deployment."""
    
    # Group components by type
    types_dict = {}
    for component in components:
        comp_type = component['type']
        if comp_type not in types_dict:
            types_dict[comp_type] = []
        types_dict[comp_type].append(component['fullName'])
    
    # Create package.xml structure
    package_xml = {
        'Package': {
            'types': [
                {
                    'name': type_name,
                    'members': members
                }
                for type_name, members in types_dict.items()
            ],
            'version': api_version
        }
    }
    
    return package_xml

# Usage
components_to_deploy = [
    {'type': 'CustomObject', 'fullName': 'MyObject__c'},
    {'type': 'ApexClass', 'fullName': 'MyController'},
    {'type': 'ApexClass', 'fullName': 'MyControllerTest'}
]

package = create_metadata_package(components_to_deploy, 'MyDeployment')

def metadata_operation_with_retry(operation_func, max_retries=3, delay=5):
    """Execute metadata operation with retry logic."""
    
    for attempt in range(max_retries):
        try:
            result = operation_func()
            
            # Check if operation succeeded
            if isinstance(result, list):
                failed_items = [r for r in result if not r['success']]
                if failed_items:
                    print(f"Attempt {attempt + 1}: {len(failed_items)} items failed")
                    if attempt < max_retries - 1:
                        time.sleep(delay)
                        continue
                    else:
                        return result
                else:
                    print(f"All items succeeded on attempt {attempt + 1}")
                    return result
            else:
                return result
                
        except Exception as e:
            print(f"Attempt {attempt + 1} failed with error: {e}")
            if attempt < max_retries - 1:
                time.sleep(delay)
            else:
                raise
    
    return result

# Usage
def create_apex_classes():
    return mdapi.ApexClass.create(apex_classes_metadata)

result = metadata_operation_with_retry(create_apex_classes, max_retries=3)

def safe_metadata_deployment(mdapi, package_path, is_production=False):
    """Deploy metadata with proper safety checks."""
    
    # Always validate first
    print("Validating deployment...")
    validation = validate_deployment(mdapi, package_path)
    
    if not validation['valid']:
        print("Validation failed - stopping deployment")
        return False
    
    # Extra safety for production
    if is_production:
        response = input("Deploy to PRODUCTION? Type 'DEPLOY' to confirm: ")
        if response != 'DEPLOY':
            print("Deployment cancelled")
            return False
    
    # Proceed with deployment
    print("Starting deployment...")
    with open(package_path, 'rb') as zipfile:
        deploy_result = mdapi.deploy(
            zipfile=zipfile,
            sandbox=not is_production,
            testLevel='RunLocalTests' if is_production else 'NoTestRun'
        )
    
    return monitor_deployment(mdapi, deploy_result['id'])

def monitor_deployment(mdapi, deployment_id):
    """Monitor deployment with detailed progress reporting."""
    
    while True:
        status = mdapi.checkDeployStatus(deployment_id)
        
        # Report progress
        total_components = status.get('numberComponentsTotal', 0)
        deployed_components = status.get('numberComponentsDeployed', 0)
        
        if total_components > 0:
            progress = (deployed_components / total_components) * 100
            print(f"Deployment progress: {progress:.1f}% ({deployed_components}/{total_components})")
        
        if status['done']:
            return status['success']
        
        time.sleep(10)

# Usage
success = safe_metadata_deployment(mdapi, '/path/to/package.zip', is_production=False)