Desenvolver aplicativos Python que usam Arquivos do Azure

Aprenda as noções básicas do Python para desenvolver aplicativos ou serviços que usam os Arquivos do Azure para armazenar dados de arquivo. Crie um aplicativo de console e aprenda a executar ações básicas com o Python e os Arquivos do Azure:

• Criar Compartilhamentos de Arquivos do Azure
• Criar diretórios
• Enumerar arquivos e diretórios em um Compartilhamento de Arquivos do Azure
• Carregar, baixar e excluir um arquivo
• Criar backups de compartilhamento de arquivos usando instantâneos

Aplica-se a

Tipo de compartilhamento de arquivos	SMB	NFS
Compartilhamentos de arquivos padrão (GPv2), LRS/ZRS
Compartilhamentos de arquivos padrão (GPv2), GRS/GZRS
Compartilhamento de arquivos premium (FileStorage), LRS/ZRS

Baixar e instalar o SDK do Armazenamento do Azure para Python

A biblioteca de clientes dos Arquivos do Azure para Python requer o Python 3.8+.

Instalar por meio de PyPi

Para instalar por meio do Índice de Pacote do Python (PyPI), digite:

pip install azure-storage-file-share

Configurar seu aplicativo para usar os Arquivos do Azure

Adicione o seguinte código próxima à parte superior de um arquivo de origem Python para usar os trechos de código neste artigo.

from azure.core.exceptions import (
    ResourceExistsError,
    ResourceNotFoundError
)

from azure.storage.fileshare import (
    ShareServiceClient,
    ShareClient,
    ShareDirectoryClient,
    ShareFileClient
)

Configurar uma conexão com os Arquivos do Azure

O ShareServiceClient permite que você trabalhe com compartilhamentos, diretórios e arquivos. O código cria um objeto ShareServiceClient usando a cadeia de conexão da conta de armazenamento:

# Create a ShareServiceClient from a connection string
service_client = ShareServiceClient.from_connection_string(connection_string)

Criar um compartilhamento de arquivos do Azure

A seguir, o código de exemplo usa um objeto ShareClient para criar o compartilhamento se ele não existir.

def create_file_share(self, connection_string, share_name):
    try:
        # Create a ShareClient from a connection string
        share_client = ShareClient.from_connection_string(
            connection_string, share_name)

        print("Creating share:", share_name)
        share_client.create_share()

    except ResourceExistsError as ex:
        print("ResourceExistsError:", ex.message)

Criar um diretório

Também é possível organizar o armazenamento colocando arquivos em subdiretórios em vez de manter todos eles no diretório raiz.

O método a seguir cria um diretório na raiz do compartilhamento de arquivo especificado usando um objeto ShareDirectoryClient.

def create_directory(self, connection_string, share_name, dir_name):
    try:
        # Create a ShareDirectoryClient from a connection string
        dir_client = ShareDirectoryClient.from_connection_string(
            connection_string, share_name, dir_name)

        print("Creating directory:", share_name + "/" + dir_name)
        dir_client.create_directory()

    except ResourceExistsError as ex:
        print("ResourceExistsError:", ex.message)

Fazer upload de um arquivo

Nesta seção, você aprenderá a carregar um arquivo do armazenamento local para os Arquivos do Azure.

O método a seguir carrega os conteúdos do arquivo especificado no diretório especificado dentro do compartilhamento de arquivo do Azure.

def upload_local_file(self, connection_string, local_file_path, share_name, dest_file_path):
    try:
        source_file = open(local_file_path, "rb")
        data = source_file.read()

        # Create a ShareFileClient from a connection string
        file_client = ShareFileClient.from_connection_string(
            connection_string, share_name, dest_file_path)

        print("Uploading to:", share_name + "/" + dest_file_path)
        file_client.upload_file(data)

    except ResourceExistsError as ex:
        print("ResourceExistsError:", ex.message)

    except ResourceNotFoundError as ex:
        print("ResourceNotFoundError:", ex.message)

Enumerar arquivos e diretórios em um Compartilhamento de Arquivos do Azure

Use o método list_directories_and_files para listar os arquivos e diretórios em um subdiretório. Esse método devolve uma paginação automática renovável. O código a seguir produz o nome de cada arquivo e subdiretório em um diretório especificado para o console.

def list_files_and_dirs(self, connection_string, share_name, dir_name):
    try:
        # Create a ShareClient from a connection string
        share_client = ShareClient.from_connection_string(
            connection_string, share_name)

        for item in list(share_client.list_directories_and_files(dir_name)):
            if item["is_directory"]:
                print("Directory:", item["name"])
            else:
                print("File:", dir_name + "/" + item["name"])

    except ResourceNotFoundError as ex:
        print("ResourceNotFoundError:", ex.message)

Baixar um arquivo

Use download_file para baixar dados de um arquivo.

O exemplo a seguir demonstra como usar o download_file para acessar o conteúdo do arquivo especificado e armazená-lo localmente com o downloaded- prefixado ao nome do arquivo.

def download_azure_file(self, connection_string, share_name, dir_name, file_name):
    try:
        # Build the remote path
        source_file_path = dir_name + "/" + file_name

        # Add a prefix to the filename to 
        # distinguish it from the uploaded file
        dest_file_name = "DOWNLOADED-" + file_name

        # Create a ShareFileClient from a connection string
        file_client = ShareFileClient.from_connection_string(
            connection_string, share_name, source_file_path)

        print("Downloading to:", dest_file_name)

        # Open a file for writing bytes on the local system
        with open(dest_file_name, "wb") as data:
            # Download the file from Azure into a stream
            stream = file_client.download_file()
            # Write the stream to the local file
            data.write(stream.readall())

    except ResourceNotFoundError as ex:
        print("ResourceNotFoundError:", ex.message)

Criar um instantâneo de compartilhamento

Você pode criar uma cópia de ponto no tempo do seu compartilhamento de arquivo inteiro.

def create_snapshot(self, connection_string, share_name):
    try:
        # Create a ShareClient from a connection string
        share_client = ShareClient.from_connection_string(
            connection_string, share_name)

        # Create a snapshot
        snapshot = share_client.create_snapshot()
        print("Created snapshot:", snapshot["snapshot"])

        # Return the snapshot time so 
        # it can be accessed later
        return snapshot["snapshot"]

    except ResourceNotFoundError as ex:
        print("ResourceNotFoundError:", ex.message)

Listar compartilhamentos e instantâneos

Você pode listar todos os instantâneos para um determinado compartilhamento.

def list_shares_snapshots(self, connection_string):
    try:
        # Create a ShareServiceClient from a connection string
        service_client = ShareServiceClient.from_connection_string(connection_string)

        # List the shares in the file service
        shares = list(service_client.list_shares(include_snapshots=True))

        for share in shares:
            if (share["snapshot"]):
                print("Share:", share["name"], "Snapshot:", share["snapshot"])
            else:
                print("Share:", share["name"])

    except ResourceNotFoundError as ex:
        print("ResourceNotFoundError:", ex.message)

Procurar instantâneo de compartilhamento

Você pode navegar em cada instantâneo de compartilhamento para recuperar arquivos e diretórios a partir desse momento.

def browse_snapshot_dir(self, connection_string, share_name, snapshot_time, dir_name):
    try:
        # Create a ShareClient from a connection string
        snapshot = ShareClient.from_connection_string(
            conn_str=connection_string, share_name=share_name, snapshot=snapshot_time)

        print("Snapshot:", snapshot_time)

        for item in list(snapshot.list_directories_and_files(dir_name)):
            if item["is_directory"]:
                print("Directory:", item["name"])
            else:
                print("File:", dir_name + "/" + item["name"])

    except ResourceNotFoundError as ex:
        print("ResourceNotFoundError:", ex.message)

Obter arquivo de instantâneo de compartilhamento

Você pode baixar um arquivo de um instantâneo de compartilhamento, que permite restaurar uma versão anterior de um arquivo.

def download_snapshot_file(self, connection_string, share_name, snapshot_time, dir_name, file_name):
    try:
        # Build the remote path
        source_file_path = dir_name + "/" + file_name

        # Add a prefix to the local filename to 
        # indicate it's a file from a snapshot
        dest_file_name = "SNAPSHOT-" + file_name

        # Create a ShareFileClient from a connection string
        snapshot_file_client = ShareFileClient.from_connection_string(
            conn_str=connection_string, share_name=share_name, 
            file_path=source_file_path, snapshot=snapshot_time)

        print("Downloading to:", dest_file_name)

        # Open a file for writing bytes on the local system
        with open(dest_file_name, "wb") as data:
            # Download the file from Azure into a stream
            stream = snapshot_file_client.download_file()
            # Write the stream to the local file
            data.write(stream.readall())

    except ResourceNotFoundError as ex:
        print("ResourceNotFoundError:", ex.message)

Excluir um único instantâneo de compartilhamento

Você pode excluir um único instantâneo de compartilhamento.

def delete_snapshot(self, connection_string, share_name, snapshot_time):
    try:
        # Create a ShareClient for a snapshot
        snapshot_client = ShareClient.from_connection_string(conn_str=connection_string, share_name=share_name, snapshot=snapshot_time)

        print("Deleting snapshot:", snapshot_time)

        # Delete the snapshot
        snapshot_client.delete_share()

    except ResourceNotFoundError as ex:
        print("ResourceNotFoundError:", ex.message)

Excluir um arquivo

Use delete_file para excluir um arquivo.

def delete_azure_file(self, connection_string, share_name, file_path):
    try:
        # Create a ShareFileClient from a connection string
        file_client = ShareFileClient.from_connection_string(
            connection_string, share_name, file_path)

        print("Deleting file:", share_name + "/" + file_path)

        # Delete the file
        file_client.delete_file()

    except ResourceNotFoundError as ex:
        print("ResourceNotFoundError:", ex.message)

Excluir compartilhamento quando existem instantâneos de compartilhamento

Use delete_share com delete_snapshots=True para excluir um compartilhamento que contém instantâneos.

def delete_share(self, connection_string, share_name):
    try:
        # Create a ShareClient from a connection string
        share_client = ShareClient.from_connection_string(
            connection_string, share_name)

        print("Deleting share:", share_name)

        # Delete the share and snapshots
        share_client.delete_share(delete_snapshots=True)

    except ResourceNotFoundError as ex:
        print("ResourceNotFoundError:", ex.message)

Próximas etapas

Agora que você aprendeu como manipular s Arquivos do Azure com o Python, siga estes links para saber mais.

• Centro de desenvolvedores do Python
• API REST de serviços de armazenamento do Azure
• SDK do Armazenamento do Microsoft Azure para Python

Para obter exemplos de código relacionados usando SDKs preteridos do Python versão 2, consulte Exemplos de código usando o Python versão 2.