Python中Azure存储账户完整结构复制报错及解决方案咨询

架构老林

2026-7-31

问题描述

场景需求：

两个Azure存储账户（源/目标）
源账户包含多个容器、虚拟文件夹、Blob等资源
需要将源账户所有内容按原结构完整复制到目标账户
若目标账户中已存在同名元素且版本比源账户旧，则覆盖该元素

编写的Python代码运行时抛出错误：AttributeError: 'BlobProperties' object has no attribute 'url'

错误原因

source_container_client.list_blobs()返回的是BlobProperties对象，该对象仅存储Blob的属性元数据（如名称、最后修改时间、大小等），并不包含url属性。原代码中直接访问source_blob.url，这是触发报错的核心原因。

解决方法

要获取源Blob的可访问URL，需通过源容器客户端构造对应的BlobClient实例，再从中提取URL。同时补充以下优化逻辑，让脚本更健壮：

过滤虚拟文件夹（BlobPrefix对象）：避免对非Blob资源执行复制操作
等待异步复制完成：start_copy_from_url是异步操作，需等待复制状态完成再继续
基础异常捕获：防止单个Blob复制失败导致整个脚本中断

修改后的完整代码：

from azure.storage.blob import BlobServiceClient, BlobPrefix
import time

# 配置存储账户连接字符串
SOURCE_CONNECTION_STRING = "你的源存储账户连接字符串"
DESTINATION_CONNECTION_STRING = "你的目标存储账户连接字符串"

# 初始化Blob服务客户端
source_blob_service_client = BlobServiceClient.from_connection_string(SOURCE_CONNECTION_STRING)
destination_blob_service_client = BlobServiceClient.from_connection_string(DESTINATION_CONNECTION_STRING)

# 遍历源账户所有容器
for source_container in source_blob_service_client.list_containers():
    container_name = source_container.name
    print(f"正在处理容器: '{container_name}'")

    # 创建目标容器（不存在则创建）
    dest_container_client = destination_blob_service_client.get_container_client(container_name)
    if not dest_container_client.exists():
        print(f"在目标账户创建容器: '{container_name}'")
        dest_container_client.create_container()

    # 获取源容器客户端
    source_container_client = source_blob_service_client.get_container_client(container_name)
    # 遍历容器内所有资源（包括虚拟文件夹）
    for item in source_container_client.list_blobs():
        # 跳过虚拟文件夹（BlobPrefix对象）
        if isinstance(item, BlobPrefix):
            continue
        
        blob_name = item.name
        dest_blob_client = dest_container_client.get_blob_client(blob_name)
        
        # 判断是否需要复制：目标不存在 或 源Blob更新
        need_copy = False
        if not dest_blob_client.exists():
            need_copy = True
        else:
            dest_props = dest_blob_client.get_blob_properties()
            if item.last_modified > dest_props.last_modified:
                need_copy = True
        
        if need_copy:
            # 构造源Blob客户端并获取URL
            source_blob_client = source_container_client.get_blob_client(blob_name)
            source_blob_url = source_blob_client.url
            
            # 启动复制操作
            copy_result = dest_blob_client.start_copy_from_url(source_blob_url)
            copy_id = copy_result['copy_id']
            
            # 等待复制完成，轮询状态
            while True:
                dest_props = dest_blob_client.get_blob_properties()
                copy_status = dest_props.copy.status
                if copy_status == 'success':
                    print(f"已复制Blob: '{blob_name}' 到容器 '{container_name}'")
                    break
                elif copy_status == 'failed':
                    print(f"复制Blob '{blob_name}' 失败: {dest_props.copy.status_description}")
                    break
                time.sleep(1)

关键修改说明

获取源Blob URL：通过source_container_client.get_blob_client(blob_name).url生成正确的Blob访问URL，替代原代码中错误的source_blob.url调用
过滤虚拟文件夹：通过isinstance(item, BlobPrefix)判断并跳过虚拟文件夹，避免无效操作
等待复制完成：添加循环轮询复制状态，确保异步复制操作完成后再处理下一个Blob
逻辑优化：整理客户端实例的获取逻辑，提升代码可读性和维护性

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