So i’ve got some classes that allow me to upload file-like/io objects to azure blob storage.
My problem here is that i want to pass a callback during the export of those objects, and this callback needs to do multiple things (hence call a higher method).

Here is the code:

from azure.storage.blob import BlobServiceClient

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 incorring."""
        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, ...):
        ...


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.

        Args:
            container (string): Container to use
            name (string): Name to use
            io (io.IOBase): File-like io object base on IOBase.
        """
        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)  # doesn't display the block function
            status.update(uploaded=sent, total=total)
            if block and callable(block):
                block.__call__(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 = { ... }
resource = Resource(...)
upload_status = UploadStatus()
strategy = provider.upload(resource, upload_status, parameters, print_progress)

Have a look at my inner function progress_callback that is the progress_hook passed to this final method https://learn.microsoft.com/en-us/python/api/azure-storage-blob/azure.storage.blob.blobclient?view=azure-python#azure-storage-blob-blobclient-upload-blob

It should call the function print_progress but that’s not the case, i don’t see any printing “PROGRESS %” appearing.

I can confirm though that file-like object is exported to azure blob storage.

Do you have any idea?