tessl/pypi-azure-mgmt-datamigration
Microsoft Azure Data Migration Client Library for Python providing comprehensive database migration management capabilities
connection-validation.mddocs/
Connection and Validation
Database connection testing and migration validation capabilities for ensuring successful migration setup and execution. These tasks verify connectivity, permissions, and migration readiness across various database platforms.
Capabilities
Source Connection Testing
Test connections to source databases across different platforms including SQL Server, MySQL, PostgreSQL, Oracle, and MongoDB.
class ConnectToSourceSqlServerTaskProperties(ProjectTaskProperties):
"""Test connection to SQL Server source database."""
input: ConnectToSourceSqlServerTaskInput
output: list[ConnectToSourceSqlServerTaskOutput]
class ConnectToSourceMySqlTaskProperties(ProjectTaskProperties):
"""Test connection to MySQL source database."""
input: ConnectToSourceMySqlTaskInput
output: list[DatabaseInfo]
class ConnectToSourcePostgreSqlSyncTaskProperties(ProjectTaskProperties):
"""Test connection to PostgreSQL source database for sync scenarios."""
input: ConnectToSourcePostgreSqlSyncTaskInput
output: list[ConnectToSourcePostgreSqlSyncTaskOutput]
class ConnectToSourceOracleSyncTaskProperties(ProjectTaskProperties):
"""Test connection to Oracle source database for sync scenarios."""
input: ConnectToSourceOracleSyncTaskInput
output: list[ConnectToSourceOracleSyncTaskOutput]
class ConnectToMongoDbTaskProperties(ProjectTaskProperties):
"""Test connection to MongoDB source database."""
input: ConnectToMongoDbTaskInput
output: list[MongoDbClusterInfo]Target Connection Testing
Test connections to target Azure database services including Azure SQL Database, SQL Managed Instance, Azure Database for MySQL, and Azure Database for PostgreSQL.
class ConnectToTargetSqlDbTaskProperties(ProjectTaskProperties):
"""Test connection to Azure SQL Database target."""
input: ConnectToTargetSqlDbTaskInput
output: list[ConnectToTargetSqlDbTaskOutput]
class ConnectToTargetSqlMITaskProperties(ProjectTaskProperties):
"""Test connection to SQL Managed Instance target."""
input: ConnectToTargetSqlMITaskInput
output: list[ConnectToTargetSqlMITaskOutput]
class ConnectToTargetAzureDbForMySqlTaskProperties(ProjectTaskProperties):
"""Test connection to Azure Database for MySQL target."""
input: ConnectToTargetAzureDbForMySqlTaskInput
output: list[ConnectToTargetAzureDbForMySqlTaskOutput]
class ConnectToTargetAzureDbForPostgreSqlSyncTaskProperties(ProjectTaskProperties):
"""Test connection to Azure Database for PostgreSQL target for sync scenarios."""
input: ConnectToTargetAzureDbForPostgreSqlSyncTaskInput
output: list[ConnectToTargetAzureDbForPostgreSqlSyncTaskOutput]Migration Validation
Validate migration inputs and configurations before executing migrations to ensure successful outcomes.
class ValidateMigrationInputSqlServerSqlDbSyncTaskProperties(ProjectTaskProperties):
"""Validate SQL Server to Azure SQL Database sync migration inputs."""
input: ValidateSyncMigrationInputSqlServerTaskInput
output: list[ValidateSyncMigrationInputSqlServerTaskOutput]
class ValidateMigrationInputSqlServerSqlMITaskProperties(ProjectTaskProperties):
"""Validate SQL Server to SQL Managed Instance migration inputs."""
input: ValidateMigrationInputSqlServerSqlMITaskInput
output: list[ValidateMigrationInputSqlServerSqlMITaskOutput]
class ValidateMongoDbTaskProperties(ProjectTaskProperties):
"""Validate MongoDB migration inputs."""
input: MongoDbMigrationSettings
output: list[MongoDbMigrationProgress]
class ValidateOracleAzureDbForPostgreSqlSyncTaskProperties(ProjectTaskProperties):
"""Validate Oracle to Azure Database for PostgreSQL sync migration inputs."""
input: MigrateOracleAzureDbPostgreSqlSyncTaskInput
output: list[ValidateOracleAzureDbPostgreSqlSyncTaskOutput]Schema and Data Discovery
Discover database schemas, tables, and objects to assist with migration planning and configuration.
class GetUserTablesSqlTaskProperties(ProjectTaskProperties):
"""Get user tables from SQL Server database."""
input: GetUserTablesSqlTaskInput
output: list[GetUserTablesSqlTaskOutput]
class GetUserTablesPostgreSqlTaskProperties(ProjectTaskProperties):
"""Get user tables from PostgreSQL database."""
input: GetUserTablesPostgreSqlTaskInput
output: list[GetUserTablesPostgreSqlTaskOutput]
class GetUserTablesOracleTaskProperties(ProjectTaskProperties):
"""Get user tables from Oracle database."""
input: GetUserTablesOracleTaskInput
output: list[GetUserTablesOracleTaskOutput]
class GetUserTablesSqlSyncTaskProperties(ProjectTaskProperties):
"""Get user tables from SQL Server for sync scenarios."""
input: GetUserTablesSqlSyncTaskInput
output: list[GetUserTablesSqlSyncTaskOutput]Usage Examples
Testing SQL Server Source Connection
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 source connection
source_connection = SqlConnectionInfo(
user_name="source_user",
password="source_password",
server_name="source-server.database.windows.net",
authentication="SqlAuthentication",
encrypt_connection=True,
trust_server_certificate=False
)
# Create connection test task
connection_input = ConnectToSourceSqlServerTaskInput(
source_connection_info=source_connection,
check_permissions_group="Default"
)
connection_properties = ConnectToSourceSqlServerTaskProperties(
input=connection_input
)
connection_task = ProjectTask(
properties=connection_properties
)
# Execute connection test
test_task = client.tasks.create_or_update(
resource_group_name="my-rg",
service_name="my-migration-service",
project_name="my-project",
task_name="test-source-connection",
parameters=connection_task
)
# Check results
task_result = client.tasks.get(
resource_group_name="my-rg",
service_name="my-migration-service",
project_name="my-project",
task_name="test-source-connection",
expand="output"
)
if task_result.properties.output:
for output in task_result.properties.output:
print(f"Database: {output.database_name}")
print(f"Server version: {output.server_version}")
print(f"Connection successful: {output.connection_succeeded}")Validating Migration Configuration
from azure.mgmt.datamigration.models import (
ValidateMigrationInputSqlServerSqlMITaskProperties,
ValidateMigrationInputSqlServerSqlMITaskInput,
MigrateSqlServerSqlMIDatabaseInput
)
# Configure validation input
validation_database = MigrateSqlServerSqlMIDatabaseInput(
name="SourceDB",
restore_database_name="TargetDB",
backup_file_share={
"path": "\\\\server\\share\\backups",
"user_name": "backup_user",
"password": "backup_password"
}
)
validation_input = ValidateMigrationInputSqlServerSqlMITaskInput(
source_connection_info=source_connection,
target_connection_info=target_connection,
selected_databases=[validation_database]
)
validation_properties = ValidateMigrationInputSqlServerSqlMITaskProperties(
input=validation_input
)
validation_task = ProjectTask(
properties=validation_properties
)
# Execute validation
validation_result = client.tasks.create_or_update(
resource_group_name="my-rg",
service_name="my-migration-service",
project_name="my-project",
task_name="validate-migration-config",
parameters=validation_task
)
# Check validation results
validation_output = client.tasks.get(
resource_group_name="my-rg",
service_name="my-migration-service",
project_name="my-project",
task_name="validate-migration-config",
expand="output"
)
if validation_output.properties.output:
for output in validation_output.properties.output:
print(f"Validation status: {output.status}")
if hasattr(output, 'validation_errors') and output.validation_errors:
for error in output.validation_errors:
print(f"Error: {error.text}")Discovering Database Schema
from azure.mgmt.datamigration.models import (
GetUserTablesSqlTaskProperties,
GetUserTablesSqlTaskInput
)
# Create schema discovery task
schema_input = GetUserTablesSqlTaskInput(
connection_info=source_connection,
selected_databases=["SourceDB"]
)
schema_properties = GetUserTablesSqlTaskProperties(
input=schema_input
)
schema_task = ProjectTask(
properties=schema_properties
)
# Execute schema discovery
schema_result = client.tasks.create_or_update(
resource_group_name="my-rg",
service_name="my-migration-service",
project_name="my-project",
task_name="discover-schema",
parameters=schema_task
)
# Review discovered tables
schema_output = client.tasks.get(
resource_group_name="my-rg",
service_name="my-migration-service",
project_name="my-project",
task_name="discover-schema",
expand="output"
)
if schema_output.properties.output:
for output in schema_output.properties.output:
print(f"Database: {output.database_name}")
if hasattr(output, 'tables'):
for table in output.tables:
print(f" Table: {table.name}")Types
class ConnectToSourceSqlServerTaskInput:
"""Input for SQL Server source connection test."""
source_connection_info: SqlConnectionInfo
check_permissions_group: str # ServerLevelPermissionsGroup enum
class ConnectToSourceSqlServerTaskOutput:
"""Output from SQL Server source connection test."""
id: str
database_name: str
server_version: str
server_brand_version: str
server_level: str
agent_jobs: list[AgentJobInfo]
logins: list[LoginInfo]
databases: list[DatabaseInfo]
class ConnectToTargetSqlDbTaskInput:
"""Input for Azure SQL Database target connection test."""
target_connection_info: SqlConnectionInfo
class ConnectToTargetSqlDbTaskOutput:
"""Output from Azure SQL Database target connection test."""
id: str
databases: list[str]
target_server_version: str
target_server_brand_version: str
class ValidateMigrationInputSqlServerSqlMITaskInput:
"""Input for SQL Managed Instance migration validation."""
source_connection_info: SqlConnectionInfo
target_connection_info: MiSqlConnectionInfo
selected_databases: list[MigrateSqlServerSqlMIDatabaseInput]
class ValidateMigrationInputSqlServerSqlMITaskOutput:
"""Output from SQL Managed Instance migration validation."""
id: str
name: str
status: str # ValidationStatus enum
validation_errors: list[ValidationError]
class GetUserTablesSqlTaskInput:
"""Input for getting SQL Server user tables."""
connection_info: SqlConnectionInfo
selected_databases: list[str]
class GetUserTablesSqlTaskOutput:
"""Output from getting SQL Server user tables."""
id: str
database_name: str
tables: list[DatabaseTable]
validation_errors: list[ValidationError]
class DatabaseTable:
"""Database table information."""
name: str
has_rows: bool
full_name: str
class ValidationError:
"""Validation error details."""
text: str
severity: str # Severity enum
class ServerLevelPermissionsGroup:
"""Server-level permission groups."""
DEFAULT = "Default"
MIGRATION_SUPPORT = "MigrationSupport"
class ValidationStatus:
"""Validation status values."""
DEFAULT = "Default"
NOT_STARTED = "NotStarted"
INITIALIZED = "Initialized"
IN_PROGRESS = "InProgress"
COMPLETED = "Completed"
COMPLETED_WITH_ISSUES = "CompletedWithIssues"
STOPPED = "Stopped"
FAILED = "Failed"tessl i tessl/pypi-azure-mgmt-datamigration@9.0.0