如何使用Python获取Azure SQL Database导入/导出作业状态

你可以通过两种Python原生方案实现导入状态查询，和PowerShell中Get-AzSqlDatabaseImportExportStatus的返回结果完全一致，可获取InProgress、Completed、Failed等标准状态值。

前置准备 #

• 保存发起bacpac导入时返回的操作状态链接（Operation Status Link），即PowerShell命令中传入-OperationstatusLink参数的完整URL
• 准备对目标Azure SQL Server有读取权限的Azure AD凭据（可使用服务主体、托管标识或本地登录的Azure CLI凭据均可）

方案1：基于Azure官方Python SDK实现 #

步骤1：安装依赖包 #

pip install azure-mgmt-sql azure-identity

步骤2：查询代码示例 #

from azure.identity import DefaultAzureCredential
from azure.mgmt.sql import SqlManagementClient

# 替换为你的Azure订阅ID、资源组名、SQL Server名、数据库名
SUBSCRIPTION_ID = "你的Azure订阅ID"
RESOURCE_GROUP_NAME = "SQL资源所在资源组名"
SERVER_NAME = "Azure SQL Server名称"
DATABASE_NAME = "目标数据库名"
# 替换为导入作业返回的操作ID（从Operation Status Link末尾截取即可）
OPERATION_ID = "导入作业操作ID"

# 初始化认证客户端
credential = DefaultAzureCredential()
sql_client = SqlManagementClient(credential, SUBSCRIPTION_ID)

# 查询导入/导出作业状态
status_result = sql_client.database_import_export_operations.get(
    resource_group_name=RESOURCE_GROUP_NAME,
    server_name=SERVER_NAME,
    database_name=DATABASE_NAME,
    operation_id=OPERATION_ID
)

# 输出核心状态信息
print(f"作业当前状态: {status_result.status}")
print(f"作业排队时间: {status_result.queued_time}")
print(f"最近更新时间: {status_result.last_modified_time}")
if status_result.error:
    print(f"作业错误信息: {status_result.error.message}")

状态值说明：返回的status字段固定返回三个枚举值：InProgress（进行中）、Succeeded（已完成，即complete状态）、Failed（导入失败）

方案2：轻量REST API调用实现（无需安装SQL管理SDK） #

如果你不想安装完整的Azure管理SDK，也可以直接模拟PowerShell命令的底层逻辑，直接请求Operation Status Link获取状态，依赖更少。

步骤1：安装轻量依赖 #

pip install requests azure-identity

步骤2：查询代码示例 #

import requests
from azure.identity import DefaultAzureCredential

# 替换为你拿到的完整Operation Status Link
OPERATION_STATUS_URL = "导入作业返回的完整OperationstatusLink地址"

# 获取Azure资源访问令牌
credential = DefaultAzureCredential()
token = credential.get_token("https://management.azure.com/.default")
headers = {
    "Authorization": f"Bearer {token.token}"
}

# 发起GET请求获取状态
resp = requests.get(OPERATION_STATUS_URL, headers=headers)
resp.raise_for_status()
status_data = resp.json()

print(f"作业当前状态: {status_data['properties']['status']}")
if status_data['properties'].get('error'):
    print(f"错误详情: {status_data['properties']['error']['message']}")

注意事项 #

• bacpac导入时长与数据库体量、目标SQL SKU性能强相关，百GB级数据库导入耗时2-5小时属于正常范围
• 轮询状态建议设置60-120秒的请求间隔，避免高频请求触发Azure API限流
• 若状态返回Failed，可直接从返回的错误字段定位原因，常见问题包括目标库存储空间不足、bacpac源文件损坏、导入账号权限不足等

内容的提问来源于stack exchange，提问作者Manish