Configuration Management

@classmethod
def get_data_center_location(cls, config: Mapping[str, Any]) -> Mapping[str, Any]:
    """
    Detect and set data center location in configuration.
    
    Parameters:
    - config: Configuration mapping containing credentials
    
    Returns:
    Updated configuration with data_center field added
    
    For API key auth: Extracts data center from apikey suffix (e.g., 'key-us10' -> 'us10')
    For OAuth auth: Makes API call to determine data center location
    """

@staticmethod
def get_oauth_data_center(access_token: str) -> str:
    """
    Get data center location for OAuth access token.
    
    Parameters:
    - access_token: OAuth access token for Mailchimp API
    
    Returns:
    Data center identifier (e.g., 'us10', 'us20')
    
    Raises:
    AirbyteTracedException: If access token is invalid or API call fails
    
    Makes request to https://login.mailchimp.com/oauth2/metadata
    to retrieve data center information embedded in token metadata.
    """

@classmethod
def migrate(cls, args: List[str], source: Source) -> None:
    """
    Orchestrate configuration migration process.
    
    Parameters:
    - args: Command-line arguments list
    - source: Airbyte source instance
    
    Process:
    1. Extract config path from arguments
    2. Read existing configuration
    3. Detect and set data center location
    4. Save updated configuration
    5. Emit control message for Airbyte platform
    
    Only performs migration if --config argument is provided.
    """

@classmethod
def modify_and_save(cls, config_path: str, source: Source, config: Mapping[str, Any]) -> Mapping[str, Any]:
    """
    Modify configuration and save to source.
    
    Parameters:
    - config_path: Path to configuration file
    - source: Airbyte source instance for config operations
    - config: Current configuration mapping
    
    Returns:
    Updated configuration with data center location
    
    Updates config with data center information and persists changes.
    """

@classmethod
def emit_control_message(cls, migrated_config: Mapping[str, Any]) -> None:
    """
    Emit control message for configuration migration.
    
    Parameters:
    - migrated_config: Updated configuration mapping
    
    Prints connector config control message in JSON format
    for Airbyte platform consumption. Required for proper
    configuration update notification.
    """

# Input configuration
config = {
    "credentials": {
        "auth_type": "apikey",
        "apikey": "your-api-key-us10"
    },
    "start_date": "2023-01-01T00:00:00.000Z"
}

# After migration
migrated_config = MigrateDataCenter.get_data_center_location(config)
# Result includes: "data_center": "us10"

# Input configuration  
config = {
    "credentials": {
        "auth_type": "oauth2.0",
        "client_id": "your-client-id",
        "client_secret": "your-client-secret",
        "access_token": "your-access-token"  
    },
    "start_date": "2023-01-01T00:00:00.000Z"
}

# Data center detected via API call
data_center = MigrateDataCenter.get_oauth_data_center(config["credentials"]["access_token"])

from source_mailchimp.config_migrations import MigrateDataCenter
from source_mailchimp import SourceMailchimp

# Load configuration
config = {"credentials": {"auth_type": "apikey", "apikey": "key-us10"}}

# Detect data center
migrated_config = MigrateDataCenter.get_data_center_location(config)
print(migrated_config["data_center"])  # Output: "us10"

import sys
from source_mailchimp import SourceMailchimp
from source_mailchimp.config_migrations import MigrateDataCenter

source = SourceMailchimp()
# Automatically handles migration before connector launch
MigrateDataCenter.migrate(sys.argv[1:], source)