Azure Blob上传时内部回调函数未触发调用问题排查

我编写了若干类用于将类文件或IO对象上传至Azure Blob Storage，希望在上传时传入回调函数执行多项操作，但实际上传成功后，回调函数并未触发（看不到预期的PROGRESS %打印内容）。

相关代码如下：

from azure.storage.blob import BlobServiceClient
import sys

class UploadStatus:

    def __init__(self, uploaded=0, total=None):
        self.uploaded = uploaded
        self.total = total
    
    def update(self, uploaded, total):
        if total is not None:
            self.total = self._normalize(total)
        if uploaded is not None:
            self.uploaded = self._normalize(uploaded)
    
    def progress(self):
        """Calculate the progress made for the upload occurring."""
        return 100.0 * self.uploaded / self.total if self.total > 0 else 0
    
    def _normalize(self, integer):
        return integer if integer else 0

class Resource:
    """A wrapper encapsulating an io-like object"""
    def __init__(self, ...):
        ...

    def with_io(self):
        # 假设这里返回一个IO对象
        ...

class AzureBlobProxy:
    def __init__(self, client):
        self.client = client

    @classmethod
    def build(cls, params: dict):
        client = BlobServiceClient(**params)
        return cls(client)

    def blob_export(self, container: str, name: str, io, block=None) -> dict:
        """Export file-like object to a Storage."""
        blob_client = self.client.get_blob_client(
            container=container,
            blob=name
        )

        return blob_client.upload_blob(
            data=io,
            progress_hook=block
        )

class AzureProvider:

    def upload(self, resource, status, params={}, block=None):
        """See BaseProvider.upload() description."""
        proxy = AzureBlobProxy.build(params)
        container, name = self._extract_container_and_name_from_params(params)

        def progress_callback(sent, total):
            print(block)  # 未打印出传入的block函数
            status.update(uploaded=sent, total=total)
            if block and callable(block):
                block(status)

        with resource.with_io() as io:
            status.update(uploaded=0, total=sys.getsizeof(io))
            proxy.blob_export(container, name, io, progress_callback)

        return self._upload_strategy()
    
    def _extract_container_and_name_from_params(self, params):
        """Return container and name for blob"""
        ...
    
    def _upload_strategy(self):
        return 'azure-blob'

def print_progress(status):
    print('PROGRESS {}%'.format(int(status.progress())))

parameters = { ... }  # 你的Azure连接参数
resource = Resource(...)
upload_status = UploadStatus()
provider = AzureProvider()
strategy = provider.upload(resource, upload_status, parameters, print_progress)

问题原因分析 #

1. 进度钩子触发阈值限制
   Azure Blob的upload_blob方法默认只有当上传的数据块达到4MB（默认分块大小）时才会触发progress_hook。如果你的上传文件/IO对象体积小于4MB，进度回调函数根本不会被调用。

2. 文件总大小获取错误
   用sys.getsizeof(io)获取的是IO对象本身的内存占用，不是文件内容的实际大小，这会导致UploadStatus的total值错误，即使回调触发，进度计算也会异常。

3. 回调传递验证缺失
   虽然代码逻辑上传递了回调，但未确认block参数是否正确传入upload方法，可能存在参数传递的隐性问题。

解决方案 #

1. 强制设置小分块大小，确保回调触发 #

修改AzureBlobProxy的blob_export方法，调用upload_blob时指定chunk_size为较小值（比如1MB），让小文件也能触发进度钩子：

def blob_export(self, container: str, name: str, io, block=None) -> dict:
    blob_client = self.client.get_blob_client(
        container=container,
        blob=name
    )

    return blob_client.upload_blob(
        data=io,
        progress_hook=block,
        chunk_size=1024 * 1024  # 设置为1MB分块
    )

2. 正确获取文件实际大小 #

替换sys.getsizeof(io)，改用IO对象的seek和tell方法获取实际内容大小（需导入os模块）：

import os

# 在AzureProvider的upload方法中修改：
with resource.with_io() as io:
    # 移动指针到文件末尾获取大小
    io.seek(0, os.SEEK_END)
    total_size = io.tell()
    # 移回指针开头，准备上传
    io.seek(0)
    status.update(uploaded=0, total=total_size)
    proxy.blob_export(container, name, io, progress_callback)

3. 验证回调参数传递 #

在AzureProvider.upload方法开头添加打印，确认block参数是否正确传入：

def upload(self, resource, status, params={}, block=None):
    print(f"Received block callback: {block}")  # 确认是否为print_progress函数
    # 后续代码...

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