如何验证Azure Blob是否成功归档上传并获取上传结果？

一、统一使用新版BlobServiceClient（替换旧版BlockBlobService） #

建议统一采用Azure Storage SDK for Python 12.x+版本的BlobServiceClient，旧版BlockBlobService已停止维护，新版功能更完善且适配归档存储层的状态校验。

二、上传阶段直接捕获成功/失败+归档状态 #

上传时无需事后单独检查，通过SDK返回结果和异常捕获直接判断上传及归档状态：

from azure.storage.blob import BlobServiceClient
import asyncio

async def upload(work_item):
    account_name = "你的存储账户名"
    account_key = "你的存储账户密钥"
    container_name = "works"
    blob_name = work_item["filename"]
    local_file_path = work_item["local_path"]
    
    try:
        # 初始化Blob服务客户端
        blob_service_client = BlobServiceClient.from_connection_string(
            f"DefaultEndpointsProtocol=https;AccountName={account_name};AccountKey={account_key};EndpointSuffix=core.windows.net"
        )
        blob_client = blob_service_client.get_blob_client(container=container_name, blob=blob_name)
        
        # 上传并直接指定归档存储层
        with open(local_file_path, "rb") as data:
            blob_properties = blob_client.upload_blob(
                data,
                overwrite=True,
                standard_blob_tier="Archive"  # 强制上传至归档层
            )
        
        # 即时验证归档状态
        if blob_properties.access_tier == "Archive":
            return True, "上传并归档成功"
        else:
            return False, "上传成功但未进入归档状态"
    
    except Exception as e:
        # 捕获所有上传异常：网络错误、权限不足、文件损坏等
        return False, f"上传失败: {str(e)}"

三、事后批量验证Blob归档状态的函数 #

如果需要对已上传的Blob做批量校验，可实现以下函数：

async def is_blob_archived(blob_name, container_name="works"):
    account_name = "你的存储账户名"
    account_key = "你的存储账户密钥"
    
    try:
        blob_service_client = BlobServiceClient.from_connection_string(
            f"DefaultEndpointsProtocol=https;AccountName={account_name};AccountKey={account_key};EndpointSuffix=core.windows.net"
        )
        blob_client = blob_service_client.get_blob_client(container=container_name, blob=blob_name)
        
        # 获取Blob属性，同时验证存在性与归档状态
        blob_properties = blob_client.get_blob_properties()
        return (
            blob_properties.exists 
            and blob_properties.access_tier == "Archive" 
            and blob_properties.access_tier_change_time is not None
        )
    
    except Exception as e:
        print(f"校验Blob {blob_name} 出错: {str(e)}")
        return False

四、任务清理逻辑 #

结合上传结果或批量校验结果，清理已成功归档的任务：

async def process_tasks(tasks):
    # 方式1：上传后即时判断并清理
    for task in tasks:
        success, msg = await upload(task)
        if success:
            print(f"任务 {task['filename']} 完成，执行清理")
            # 这里写删除任务的逻辑：比如删除本地文件、标记数据库状态等
            # os.remove(task['local_path'])
        else:
            print(f"任务 {task['filename']} 失败: {msg}，保留任务")
    
    # 方式2：批量校验已上传任务并清理
    # for task in tasks:
    #     if await is_blob_archived(task['filename']):
    #         # 执行清理逻辑
    #         pass
    #     else:
    #         # 保留任务，可触发重试上传
    #         pass

关键注意事项 #

• 归档状态延迟：若将已存在的Blob从热/冷层转换至归档层，会短暂处于rehydrate_pending状态，此时需等待转换完成再校验；直接上传至归档层的Blob会立即进入Archive状态。
• 异常覆盖：上传时需捕获所有可能的异常，包括网络中断、权限不足、本地文件缺失等，这些场景均属于上传失败，任务需保留。
• SDK版本要求：确保安装新版SDK：pip install azure-storage-blob>=12.0.0

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