skip to Main Content

Below steps are configured.

1] Azure managed identity has been enabled for the vm
2] Access granted to Azure blob storage from the vm

I wanted to clarify what will be the next steps to access the blob and list down the directories.

Do we need to configure anything on the vm to use the credentials? is CLI mandatory if i just want to access directly via a python code?

Error message received :


Listing directories in container: optitex
ImdsCredential.get_token_info failed: ManagedIdentityCredential authentication unavailable.
ManagedIdentityCredential.get_token_info failed: ManagedIdentityCredential authentication unavailable.

Code I am using:

from azure.storage.blob import BlobServiceClient
from azure.core.exceptions import ResourceNotFoundError

# Function to list directories in a container
def list_directories_in_blob_container(storage_account_name, container_name):
    try:
        # Create the BlobServiceClient using DefaultAzureCredential (for Managed Identity)
        # If DefaultAzureCredential fails, fall back to AzureCliCredential
        try:
            print('********')
            credential = ManagedIdentityCredential()
        except Exception as e:
            print(f"DefaultAzureCredential failed: {e}")
            print("Falling back to Default Credential")
            credential = DefaultAzureCredential()

        blob_service_client = BlobServiceClient(account_url=f"https://{storage_account_name}.blob.core.windows.net", credential=credential)

        # Get the container client
        container_client = blob_service_client.get_container_client(container_name)

        # List all blobs in the container and check for directories
        print(f"Listing directories in container: {container_name}")
        blob_list = container_client.walk_blobs()

        directories = set()

        for blob in blob_list:
            blob_path_parts = blob.name.split('/')
            if len(blob_path_parts) > 1:
                directories.add(blob_path_parts[0])

        # Display found directories
        if directories:
            print("Directories found:")
            for directory in sorted(directories):
                print(f" - {directory}")
        else:
            print("No directories found.")

    except ResourceNotFoundError:
        print(f"Container '{container_name}' not found in the storage account '{storage_account_name}'.")
    except Exception as ex:
        print(f"An error occurred: {ex}")

# Example usage
if __name__ == "__main__":
    storage_account_name = "xxx"
    container_name = "xxx"

    list_directories_in_blob_container(storage_account_name, container_name)```


2

Answers


  1. Have you tried below to initializate credential?

    from azure.identity import ManagedIdentityCredential
    
    credential = ManagedIdentityCredential()
    
    # Can also specify a client ID of a user-assigned managed identity
    credential = ManagedIdentityCredential(
       client_id="<client_id>",
    )
    

    reference

    Login or Signup to reply.
  2. According to this SO-Answer by Allen Wu.

    Both system and user managed identity is not supported with ManagedIdentityCredential in the local environment.

    In case you want to use a user-asigned managed identity with the DefaultAzureCredential when deployed to Azure, specify the clientId.

    I tried with below modified code it worked in my environment and also assign Storage blob data contributor role to the identity.

    Code:

    from azure.storage.blob import BlobServiceClient
    from azure.core.exceptions import ResourceNotFoundError
    from azure.identity import DefaultAzureCredential
    
    
    def list_directories_in_blob_container(storage_account_name, container_name):
        try:
            client_id=""
            credential=DefaultAzureCredential(managed_identity_client_id=client_id)
            blob_service_client = BlobServiceClient(account_url=f"https://{storage_account_name}.blob.core.windows.net", credential=credential)
            container_client = blob_service_client.get_container_client(container_name)
            print(f"Listing directories in container: {container_name}")
            blob_list = container_client.walk_blobs()
    
            directories = set()
            for blob in blob_list:
                blob_path_parts = blob.name.split('/')
                if len(blob_path_parts) > 1:
                    directories.add(blob_path_parts[0])
    
            if directories:
                print("Directories found:")
                for directory in sorted(directories):
                    print(f" - {directory}")
            else:
                print("No directories found.")
    
        except ResourceNotFoundError:
            print(f"Container '{container_name}' not found in the storage account '{storage_account_name}'.")
        except Exception as ex:
            print(f"An error occurred: {ex}")
    
    # Example usage
    if __name__ == "__main__":
        storage_account_name = "venkat326123"
        container_name = "data"
    
        list_directories_in_blob_container(storage_account_name, container_name)
    

    Output:

    Listing directories in container: data
    Directories found:
     - samplehub1
     - test
    

    enter image description here

    Reference:
    Azure Identity client library for Python | Microsoft Learn

    Login or Signup to reply.
Please signup or login to give your own answer.
Back To Top
Search