Conectando ao Azure Blob Storage com Python para Listar Arquivos em um Diretório

Neste artigo, vou mostrar como você pode se conectar ao Azure Blob Storage usando o Python e listar arquivos em um diretório específico. Vamos utilizar bibliotecas como azure.storage.blob e mssparkutils para interagir com o serviço de armazenamento e obter segredos do Azure Key Vault.

Configurações Iniciais
Primeiramente, precisamos definir algumas configurações iniciais, como o nome do container, o caminho relativo dos arquivos e o serviço vinculado que vamos usar.

`from azure.storage.blob import BlobServiceClient
import json

# Configurações iniciais
blob_container_name = 'container_aqui'  # seu nome do container
blob_relative_path_enriched = 'caminho_aqui'  # caminho com barra no final
linked_service_enriched = 'Linked_service_aqui'  # seu linked service
`

Obtendo Propriedades do Key Vault
Vamos utilizar o mssparkutils para obter as propriedades do Key Vault e a chave de acesso necessária.

`# Obtendo propriedades do Key Vault
ls_keyvault = mssparkutils.credentials.getPropertiesAll('Linked_service_aqui')
converter_dic_kv = json.loads(ls_keyvault)

# Coletando o endpoint do Key Vault
end_point_kv = (converter_dic_kv['Endpoint'].split("/"))[2]
kv_name = (end_point_kv.split("."))[0]

# Obtendo a chave de acesso do Key Vault
access_key = mssparkutils.credentials.getSecret(kv_name, 'seu_storage_aqui', 'Linked_service_aqui')
`

Configurando o Cliente do Blob Storage
Agora que temos a chave de acesso, vamos configurar o cliente do Blob Storage.

# Obtendo propriedades do linked service
ls_enriched = mssparkutils.credentials.getPropertiesAll(Linked_service_aqui_enriched)
converter_dic_enriched = json.loads(ls_enriched)

# Coletando o endpoint do linked service
end_point_enriched = (converter_dic_enriched['Endpoint'].split("/"))[2]

# Configurando o cliente do Blob Storage
storage_account = end_point_enriched.split(".")[0]
print(f"Storage Account: {storage_account}")
print(f"Access Key: {access_key[:4]}...{access_key[-4:]}")  # Mostrar apenas início e fim da chave por segurança

blob_service_client = BlobServiceClient(account_url=f"https://{storage_account}.blob.core.windows.net", credential=access_key)
container_client = blob_service_client.get_container_client(blob_container_name)

Função para Listar Arquivos
Vamos criar uma função para listar os nomes dos arquivos em um diretório específico no Blob Storage.

def list_files_in_directory(container_client, directory_path):
    """ Função para listar os nomes dos arquivos em um diretório específico """
    try:
        print(f"Tentando acessar o container: {container_client.container_name}")
        blob_list = container_client.list_blobs(name_starts_with=directory_path)
        file_names = [blob.name for blob in blob_list]
        if not file_names:
            print(f"Nenhum arquivo encontrado em {directory_path}.")
        return file_names
    except Exception as e:
        print(f"Erro ao listar arquivos no diretório especificado: {e}")
        return []

Listando os Arquivos
Agora podemos usar a função list_files_in_directory para listar os arquivos no diretório especificado.

# Listando os nomes dos arquivos no diretório especificado
print(f"Tentando listar arquivos no diretório: {blob_relative_path_enriched}")

file_names = list_files_in_directory(container_client, blob_relative_path_enriched)

# Verificando se a lista de arquivos não está vazia
if not file_names:
    print("Nenhum arquivo encontrado no diretório especificado.")
else:
    # Exibindo os nomes dos arquivos
    print(f"Arquivos encontrados no diretório {blob_relative_path_enriched}:")
    for file_name in file_names:
        print(file_name)

print("Processo finalizado.")

Com isto podemos listar todos os dados dentro de um blob utilizando um notebook com Synapase junto a Microsoft Azure.

Agradecimento Especial à André Luiz dos Santos Junior pelo apoio na solução : https://www.linkedin.com/in/andrelsjunior/

Agradecimento Especial à Adilton Costa Anna pelo apoio na solução : https://www.linkedin.com/in/adiltoncantos/