基于Python实现Azure Blob Storage文件断点续传

要实现中断后从中断处恢复上传且不覆盖现有内容，得利用Azure Blob存储的Block Blob分块上传机制——每个上传的块有唯一ID，上传后可查询已上传的块列表，跳过已上传部分继续上传剩余内容。以下是具体实现方案：

核心逻辑 #

• 先检查目标Blob是否存在，若存在则拉取所有已上传的块，计算已上传的总字节数
• 将本地文件的读取指针定位到已上传字节的末尾，从该位置开始读取剩余内容分块上传
• 每个块用唯一ID标识，全部上传完成后提交块列表，完成Blob的拼接

完整代码实现 #

from azure.storage.blob import BlobServiceClient, BlobBlock
import os

azure_container = "dummy-container"
file_path = "test.txt"
chunk_size = 4 * 1024 * 1024  # 4MB块大小
azure_connection_string = "你的Azure存储连接字符串"

# 初始化Blob客户端
blob_service_client = BlobServiceClient.from_connection_string(azure_connection_string)
blob_client = blob_service_client.get_blob_client(container=azure_container, blob="testingfile.txt")

def resume_upload():
    uploaded_blocks = []
    uploaded_total_size = 0

    # 检查目标Blob是否存在，读取已上传的块信息
    if blob_client.exists():
        # 获取所有已提交和未提交的块列表
        block_list = blob_client.get_block_list(block_list_type="all")
        uploaded_blocks = [block.name for block in block_list.committed_blocks + block_list.uncommitted_blocks]
        # 计算已上传的总字节数
        uploaded_total_size = sum(len(block) for block in block_list.committed_blocks + block_list.uncommitted_blocks)
        print(f"检测到已上传 {uploaded_total_size} 字节，将从该位置续传")

    # 打开本地文件，定位到已上传内容的末尾
    with open(file_path, 'rb') as datax:
        datax.seek(uploaded_total_size)
        current_block_index = len(uploaded_blocks)

        while True:
            chunk_data = datax.read(chunk_size)
            if not chunk_data:
                break  # 文件读取完毕

            # 生成符合Azure要求的唯一块ID（Base64兼容格式）
            block_id = str(current_block_index).zfill(6).encode('utf-8').hex()
            current_block_index += 1

            # 上传单个块
            try:
                blob_client.stage_block(block_id=block_id, data=chunk_data)
                uploaded_blocks.append(block_id)
                uploaded_total_size += len(chunk_data)
                print(f"已完成块 {current_block_index-1} 上传，累计上传 {uploaded_total_size} 字节")
            except Exception as e:
                print(f"块上传失败: {str(e)}")
                # 实际场景可添加重试逻辑，这里直接抛出异常终止
                raise

        # 所有块上传完成后，提交块列表生成完整Blob
        if uploaded_blocks:
            blob_client.commit_block_list(block_list=uploaded_blocks)
            print("文件上传/续传完成")

if __name__ == "__main__":
    resume_upload()

关键细节说明 #

• 块ID规则：Azure要求块ID为Base64编码的字符串，这里用填充后的数字转十六进制，保证唯一性和格式合规
• 已上传块校验：通过get_block_list获取所有已上传的块，避免重复上传已成功的内容
• 文件指针定位：用seek(uploaded_total_size)直接跳到已上传内容的末尾，无需重新读取已上传部分
• 异常容错：单个块上传失败时可添加重试逻辑（比如3次重试），应对临时网络波动
• 提交块列表：必须调用commit_block_list才能将零散的块拼接成完整的Blob，否则无法正常访问文件

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