CtrlK

Community Documentation Log in Get started

tessl/pypi-azure-mgmt-datamigration

Microsoft Azure Data Migration Client Library for Python providing comprehensive database migration management capabilities

Overview

Eval results

Files

Migration Tasks

Name: tessl/pypi-azure-mgmt-datamigration
Author: tessl

Execution and monitoring of specific migration tasks within projects. Tasks perform the actual work of migration, including connection testing, schema migration, data migration, and validation across various database platforms.

Capabilities

Task Lifecycle Management

Create, update, delete, and retrieve migration tasks with comprehensive configuration for different migration scenarios.

def create_or_update(
    group_name: str,
    service_name: str,
    project_name: str,
    task_name: str,
    parameters: ProjectTask,
    **kwargs
) -> ProjectTask:
    """
    Create or update a migration task.
    
    Parameters:
    - group_name: Name of the resource group
    - service_name: Name of the Data Migration Service
    - project_name: Name of the project
    - task_name: Name of the task
    - parameters: Task configuration parameters
    
    Returns:
    ProjectTask instance details
    """

def get(
    group_name: str,
    service_name: str,
    project_name: str,
    task_name: str,
    expand: str = None,
    **kwargs
) -> ProjectTask:
    """
    Get details of a migration task.
    
    Parameters:
    - group_name: Name of the resource group
    - service_name: Name of the Data Migration Service
    - project_name: Name of the project
    - task_name: Name of the task
    - expand: Expand the response with additional details
    
    Returns:
    ProjectTask instance details
    """

def delete(
    group_name: str,
    service_name: str,
    project_name: str,
    task_name: str,
    delete_running_tasks: bool = None,
    **kwargs
) -> None:
    """
    Delete a migration task.
    
    Parameters:
    - group_name: Name of the resource group
    - service_name: Name of the Data Migration Service
    - project_name: Name of the project
    - task_name: Name of the task
    - delete_running_tasks: Whether to delete if running
    """

def update(
    group_name: str,
    service_name: str,
    project_name: str,
    task_name: str,
    parameters: ProjectTask,
    **kwargs
) -> ProjectTask:
    """
    Update an existing migration task.
    
    Parameters:
    - group_name: Name of the resource group
    - service_name: Name of the Data Migration Service
    - project_name: Name of the project
    - task_name: Name of the task
    - parameters: Updated task parameters
    
    Returns:
    Updated task details
    """

Task Control Operations

Cancel running tasks and execute commands on tasks for operational control during migration execution.

def cancel(
    group_name: str,
    service_name: str,
    project_name: str,
    task_name: str,
    **kwargs
) -> ProjectTask:
    """
    Cancel a running migration task.
    
    Parameters:
    - group_name: Name of the resource group
    - service_name: Name of the Data Migration Service
    - project_name: Name of the project
    - task_name: Name of the task
    
    Returns:
    Updated task details
    """

def command(
    group_name: str,
    service_name: str,
    project_name: str,
    task_name: str,
    parameters: CommandProperties,
    **kwargs
) -> CommandProperties:
    """
    Execute a command on a migration task.
    
    Parameters:
    - group_name: Name of the resource group
    - service_name: Name of the Data Migration Service
    - project_name: Name of the project
    - task_name: Name of the task
    - parameters: Command parameters
    
    Returns:
    Command execution results
    """

Task Discovery

List migration tasks within a project for monitoring and management purposes.

def list(
    group_name: str,
    service_name: str,
    project_name: str,
    task_type: str = None,
    **kwargs
) -> ItemPaged[ProjectTask]:
    """
    List migration tasks in a project.
    
    Parameters:
    - group_name: Name of the resource group
    - service_name: Name of the Data Migration Service
    - project_name: Name of the project
    - task_type: Filter by task type
    
    Returns:
    Paginated list of tasks
    """

Usage Examples

Creating a SQL Server Connection Test Task

from azure.mgmt.datamigration import DataMigrationManagementClient
from azure.mgmt.datamigration.models import (
    ProjectTask, ConnectToSourceSqlServerTaskProperties,
    ConnectToSourceSqlServerTaskInput, SqlConnectionInfo
)
from azure.identity import DefaultAzureCredential

credential = DefaultAzureCredential()
client = DataMigrationManagementClient(credential, "subscription-id")

# Configure connection information
source_connection = SqlConnectionInfo(
    user_name="test_user",
    password="test_password", 
    server_name="source-server.database.windows.net",
    authentication="SqlAuthentication",
    encrypt_connection=True
)

# Create connection test task
task_input = ConnectToSourceSqlServerTaskInput(
    source_connection_info=source_connection,
    check_permissions_group="Default"
)

task_properties = ConnectToSourceSqlServerTaskProperties(
    input=task_input
)

task = ProjectTask(
    properties=task_properties
)

# Create the task
created_task = client.tasks.create_or_update(
    group_name="my-rg",
    service_name="my-migration-service",
    project_name="my-project",
    task_name="connect-source-test",
    parameters=task
)

print(f"Task created: {created_task.name}")
print(f"Task type: {created_task.properties.task_type}")

Creating a Data Migration Task

from azure.mgmt.datamigration.models import (
    MigrateSqlServerSqlDbTaskProperties,
    MigrateSqlServerSqlDbTaskInput,
    MigrateSqlServerSqlDbDatabaseInput
)

# Configure database migration input
database_input = MigrateSqlServerSqlDbDatabaseInput(
    name="SourceDB",
    target_database_name="TargetDB",
    make_source_db_read_only=False,
    table_map={
        "dbo.Users": "dbo.Users",
        "dbo.Orders": "dbo.Orders"
    }
)

# Create migration task input
migration_input = MigrateSqlServerSqlDbTaskInput(
    source_connection_info=source_connection,
    target_connection_info=target_connection,
    selected_databases=[database_input],
    validation_options={
        "enable_schema_validation": True,
        "enable_data_integrity_validation": True,
        "enable_query_analysis_validation": False
    }
)

# Create migration task properties
migration_properties = MigrateSqlServerSqlDbTaskProperties(
    input=migration_input
)

migration_task = ProjectTask(
    properties=migration_properties
)

# Create the migration task
migration_created_task = client.tasks.create_or_update(
    group_name="my-rg",
    service_name="my-migration-service", 
    project_name="my-project",
    task_name="migrate-data-task",
    parameters=migration_task
)

Monitoring Task Progress

# Get task details with expanded information
task_details = client.tasks.get(
    group_name="my-rg",
    service_name="my-migration-service",
    project_name="my-project", 
    task_name="migrate-data-task",
    expand="output"
)

print(f"Task state: {task_details.state}")
print(f"Task type: {task_details.properties.task_type}")

# Check if task has output
if hasattr(task_details.properties, 'output') and task_details.properties.output:
    for output in task_details.properties.output:
        if hasattr(output, 'progress'):
            print(f"Progress: {output.progress}%")
        if hasattr(output, 'status'):
            print(f"Status: {output.status}")

# Cancel task if needed
if task_details.state == "Running":
    cancelled_task = client.tasks.cancel(
        group_name="my-rg",
        service_name="my-migration-service",
        project_name="my-project",
        task_name="migrate-data-task"
    )
    print(f"Task cancelled: {cancelled_task.state}")

Managing Migration Commands

from azure.mgmt.datamigration.models import (
    MigrateSyncCompleteCommandProperties,
    MigrateSyncCompleteCommandInput
)

# Complete a sync migration
complete_command_input = MigrateSyncCompleteCommandInput(
    database_name="TargetDB"
)

complete_command = MigrateSyncCompleteCommandProperties(
    input=complete_command_input
)

# Execute the complete command
command_result = client.tasks.command(
    group_name="my-rg",
    service_name="my-migration-service",
    project_name="my-project",
    task_name="sync-migration-task",
    parameters=complete_command
)

print(f"Command executed: {command_result.command_type}")

Types

class ProjectTask:
    """Migration task configuration and state."""
    name: str
    type: str
    etag: str
    properties: ProjectTaskProperties
    state: str  # TaskState enum

class ProjectTaskProperties:
    """Base class for task properties."""
    task_type: str
    errors: list[ODataError]
    state: str  # TaskState enum
    commands: list[CommandProperties]

class ConnectToSourceSqlServerTaskProperties(ProjectTaskProperties):
    """SQL Server source connection task properties."""
    input: ConnectToSourceSqlServerTaskInput
    output: list[ConnectToSourceSqlServerTaskOutput]

class ConnectToSourceSqlServerTaskInput:
    """Input for SQL Server connection task."""
    source_connection_info: SqlConnectionInfo
    check_permissions_group: str  # ServerLevelPermissionsGroup enum

class MigrateSqlServerSqlDbTaskProperties(ProjectTaskProperties):
    """SQL Server to SQL Database migration task properties."""
    input: MigrateSqlServerSqlDbTaskInput
    output: list[MigrateSqlServerSqlDbTaskOutput]

class MigrateSqlServerSqlDbTaskInput:
    """Input for SQL Server to SQL Database migration."""
    source_connection_info: SqlConnectionInfo
    target_connection_info: SqlConnectionInfo
    selected_databases: list[MigrateSqlServerSqlDbDatabaseInput]
    validation_options: MigrationValidationOptions

class MigrateSqlServerSqlDbDatabaseInput:
    """Database input for SQL Server migration."""
    name: str
    target_database_name: str
    make_source_db_read_only: bool
    table_map: dict[str, str]

class MigrationValidationOptions:
    """Validation options for migration."""
    enable_schema_validation: bool
    enable_data_integrity_validation: bool
    enable_query_analysis_validation: bool

class TaskState:
    """Task execution states."""
    UNKNOWN = "Unknown"
    QUEUED = "Queued"
    RUNNING = "Running"
    CANCELED = "Canceled"
    SUCCEEDED = "Succeeded"  
    FAILED = "Failed"
    FAILED_INPUT_VALIDATION = "FailedInputValidation"
    FAULTED = "Faulted"

class CommandProperties:
    """Base class for task commands."""
    command_type: str
    errors: list[ODataError]
    state: str  # CommandState enum

class MigrateSyncCompleteCommandProperties(CommandProperties):
    """Complete sync migration command."""
    input: MigrateSyncCompleteCommandInput
    output: MigrateSyncCompleteCommandOutput

class MigrateSyncCompleteCommandInput:
    """Input for sync complete command."""
    database_name: str

tessl i tessl/pypi-azure-mgmt-datamigration@9.0.0

docs

connection-validation.md