Python实现Azure Blob存储A到B新增文件复制及SDK兼容问题咨询

核心思路 #

放弃已弃用的旧版BlockBlobService（属于v2 SDK），改用官方推荐的v12+版本azure-storage-blob包，通过BlobServiceClient统一操作两个存储账户，彻底解决新旧SDK不兼容的问题，同时实现仅复制目标存储中不存在的新文件的需求。

实现步骤 #

1. 安装依赖
   确保安装最新版Azure Blob存储SDK：

pip install azure-storage-blob

2. 代码实现
   以下是完整的增量复制逻辑，直接通过Azure后台完成复制（无需下载到本地，效率更高）：

from azure.storage.blob import BlobServiceClient, ContainerClient

# 替换为你的存储账户连接字符串
SOURCE_CONN_STR = "存储A的连接字符串"
TARGET_CONN_STR = "存储B的连接字符串"

# 替换为对应的容器名称
SOURCE_CONTAINER_NAME = "源容器名称"
TARGET_CONTAINER_NAME = "目标容器名称"

def copy_new_blobs():
    # 初始化源、目标存储的Blob服务客户端
    source_blob_service = BlobServiceClient.from_connection_string(SOURCE_CONN_STR)
    target_blob_service = BlobServiceClient.from_connection_string(TARGET_CONN_STR)

    # 获取容器客户端
    source_container = source_blob_service.get_container_client(SOURCE_CONTAINER_NAME)
    target_container = target_blob_service.get_container_client(TARGET_CONTAINER_NAME)

    # 预加载目标容器所有Blob名称，减少重复查询次数
    target_blob_names = {blob.name for blob in target_container.list_blobs()}

    # 遍历源容器内所有文件
    for source_blob in source_container.list_blobs():
        blob_name = source_blob.name
        # 检查目标容器是否不存在该文件
        if blob_name not in target_blob_names:
            # 获取源Blob的访问路径
            source_blob_url = source_container.get_blob_client(blob_name).url
            # 启动后台复制任务
            target_blob_client = target_container.get_blob_client(blob_name)
            target_blob_client.start_copy_from_url(source_blob_url)
            print(f"已复制新文件: {blob_name}")

if __name__ == "__main__":
    copy_new_blobs()

3. 私有存储账户的权限处理
   如果存储账户是私有的，直接使用Blob URL会无权限访问，需要为源Blob生成SAS令牌，添加以下辅助函数即可：

from azure.storage.blob import generate_blob_sas, BlobSasPermissions
from datetime import datetime, timedelta

def generate_sas_token(container_client: ContainerClient, blob_name: str):
    sas_token = generate_blob_sas(
        account_name=container_client.account_name,
        container_name=container_client.container_name,
        blob_name=blob_name,
        account_key=container_client.credential.account_key,
        permission=BlobSasPermissions(read=True),
        expiry=datetime.utcnow() + timedelta(hours=1)  # 有效期可按需调整
    )
    return f"{container_client.get_blob_client(blob_name).url}?{sas_token}"

复制时将source_blob_url替换为generate_sas_token(source_container, blob_name)的返回值即可。

注意事项 #

BlockBlobService属于旧版v2 SDK，微软已停止维护，其API设计与v12+版本完全不同，混用必然导致兼容问题。统一使用v12+ SDK不仅能解决兼容问题，还能获得更好的性能和官方持续支持的新特性。

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