Azure Storageエミュレーター、Azuriteを試す

これは、なにをしたくて書いたもの？

Azure Storageを、ローカルで動かすためのエミュレーターがあるようです。

開発とテストに Azure ストレージ エミュレーターを使用する (非推奨) | Microsoft Docs

ですが、こちらはあまり開発されていないうえに、Windowsでのみ動作するようです。

Azure Storage エミュレーターは現在、あまり開発されていません。 Azurite が今後のストレージ エミュレーター プラットフォームです。
ストレージ エミュレーターは、現在、Windows でのみ実行されます。

代わりにAzuriteというものを使った方が良さそうなので、今回こちらを試してみたいと思います。

Azurite

Azuriteは、オープンソースのAzure Blob StorageおよびAzure Queue Storageのエミュレーターです。

ローカルでの Azure Storage の開発に Azurite エミュレーターを使用する | Microsoft Docs

GitHub - Azure/Azurite: A lightweight server clone of Azure Storage that simulates most of the commands supported by it with minimal dependencies

Windows、Linux、macOSと、クロスプラットフォームで動作します。

インストール方法は、Visual Studio Extension、npm、Dockerイメージの3つがあります。

Azurite - Visual Studio Marketplace

azurite - npm

DockerHub / Azurite

今回は、npmモジュールとしてインストールしてみます。

環境

今回の環境は、こちらです。

$ npm -v
6.14.11


$ node -v
v14.16.0

Azuriteをインストールする

こちらの手順に沿って、Azuriteをnpmでインストールしてみます。

NPM を使用して Azurite をインストールして実行する

とはいえ、グローバルに入れるのは抵抗感があったので、-gオプションは付与しませんが。

$ npm i azurite

バージョン確認。

$ npx azurite -v
3.11.0

ヘルプを確認。

$ npx azurite -h
  Usage: azurite [options] [command]
  
  Commands:
    help     Display help
    version  Display version
  
  Options:
    -, --blobHost [value]     Optional. Customize listening address for blob (defaults to "127.0.0.1")
    -, --blobPort <n>         Optional. Customize listening port for blob (defaults to 10000)
    -, --cert                 Optional. Path to certificate file
    -d, --debug               Optional. Enable debug log by providing a valid local file path as log destination
    -h, --help                Output usage information
    -, --key                  Optional. Path to certificate key .pem file
    -l, --location [value]    Optional. Use an existing folder as workspace path, default is current working directory (defaults to "/path/to/current-dir")
    -L, --loose               Optional. Enable loose mode which ignores unsupported headers and parameters
    -, --oauth                Optional. OAuth level. Candidate values: "basic"
    -, --pwd                  Optional. Password for .pfx file
    -, --queueHost [value]    Optional. Customize listening address for queue (defaults to "127.0.0.1")
    -, --queuePort <n>        Optional. Customize listening port for queue (defaults to 10001)
    -s, --silent              Optional. Disable access log displayed in console
    -, --skipApiVersionCheck  Optional. Skip the request API version check, request with all Api versions will be allowed
    -v, --version             Output the version number
  

Azuriteの起動方法は、こちら。

Azurite をコマンド ラインから実行する

このあたりのオプションが重要そうですね。バインドするアドレスやポート、データの保存ディレクトリです。

    -, --blobHost [value]     Optional. Customize listening address for blob (defaults to "127.0.0.1")
    -, --blobPort <n>         Optional. Customize listening port for blob (defaults to 10000)


    -l, --location [value]    Optional. Use an existing folder as workspace path, default is current working directory (defaults to "/path/to/current-directory")


    -, --queueHost [value]    Optional. Customize listening address for queue (defaults to "127.0.0.1")
    -, --queuePort <n>        Optional. Customize listening port for queue (defaults to 10001)

データを保存するディレクトリを作成。

$ mkdir data

今回は、-lオプションを使用して起動することにしました。

$ npx azurite -l data
Azurite Blob service is starting at http://127.0.0.1:10000
Azurite Blob service is successfully listening at http://127.0.0.1:10000
Azurite Queue service is starting at http://127.0.0.1:10001
Azurite Queue service is successfully listening at http://127.0.0.1:10001

これで、準備は完了です。

Pythonから接続してみる

AzuriteでAzure Storageのエミュレーターが起動したので、プログラムからアクセスしてみましょう。

クイックスタート: Azure Blob Storage ライブラリ v12 - Python | Microsoft Docs

リファレンスは、こちら。

Azure Storage client libraries for Python | Microsoft Docs

Azure Storage Blobs client library for Python | Microsoft Docs

今回は、Pythonで利用してみます。

Pythonに関する環境情報は、こちら。

$ python3 -V
Python 3.8.5


$ pip3 -V
pip 20.0.2 from /usr/lib/python3/dist-packages/pip (python 3.8)

接続に関する情報は、このあたりを見たら良さそうです。

Azurite V3 / Usage with Azure Storage SDKs or Tools

デフォルトのストレージアカウントの情報。

Default Storage Account

• アカウント名 … devstoreaccount1
• アカウントキー … Eby8vdM02xNOcqFlqUwJPLlmEtlCDXJ1OUzFT50uSRZ6IFsuFq2UVErCz4I6tq/K1SZFPTOtr/KBHBeksoGMGw==

環境変数でカスタマイズして設定することもできるようです。

Customized Storage Accounts & Keys

接続文字列は、こちら。

Connection Strings

では、プログラムを作成していきます。

まずは、パッケージをインストールしましょう。

クイック スタート:Python v12 SDK で BLOB を管理する / パッケージをインストールする

$ pip3 install azure-storage-blob==12.8.0

動作確認は、テストコードで行うことにします。pytestもインストールしましょう。

$ pip3 install pytest==6.2.2

あとは、こちらのコード例を見ながら進めていきます。

クイック スタート:Python v12 SDK で BLOB を管理する / コード例

テストコードの雛形を用意。

tests/test_azurite_access.py

from azure.core.paging import ItemPaged
from azure.storage.blob import BlobServiceClient, BlobClient, ContainerClient, BlobProperties

# ここに、テストコードを書く！

接続文字列を定義。

connection_string: str = 'DefaultEndpointsProtocol=http;AccountName=devstoreaccount1;AccountKey=Eby8vdM02xNOcqFlqUwJPLlmEtlCDXJ1OUzFT50uSRZ6IFsuFq2UVErCz4I6tq/K1SZFPTOtr/KBHBeksoGMGw==;BlobEndpoint=http://127.0.0.1:10000/devstoreaccount1;QueueEndpoint=http://127.0.0.1:10001/devstoreaccount1;'

まずは、Blobサービスおよびコンテナへアクセスしてみます。

def test_connect_container():
    blob_service_client: BlobServiceClient = BlobServiceClient.from_connection_string(connection_string)

    container_name: str = 'my-blob-container'

    container_client: ContainerClient = blob_service_client.create_container(container_name)

    try:
        list_blobs: ItemPaged = container_client.list_blobs()

        blobs: list = []

        for blob in list_blobs:
            blob: BlobProperties = blob
            blobs.append(blob.name)

        assert len(blobs) == 0
    finally:
        container_client.delete_container()
        container_client.close()
        blob_service_client.close()

Blobサービスがアカウントレベルのリソース、コンテナはディレクトリに相当する概念のようです。

Blob (オブジェクト) Storage の概要 - Azure Storage | Microsoft Docs

クイック スタート:Python v12 SDK で BLOB を管理する / コンテナーを作成する

    blob_service_client: BlobServiceClient = BlobServiceClient.from_connection_string(connection_string)

    container_name: str = 'my-blob-container'

    container_client: ContainerClient = blob_service_client.create_container(container_name)

まだなにも作成していないので、コンテナ内は空です。

        list_blobs: ItemPaged = container_client.list_blobs()

        blobs: list = []

        for blob in list_blobs:
            blob: BlobProperties = blob
            blobs.append(blob.name)

        assert len(blobs) == 0

最後に、コンテナを削除して各種クライアントをクローズします。

クイック スタート:Python v12 SDK で BLOB を管理する / コンテナーを削除する

    finally:
        container_client.delete_container()
        container_client.close()
        blob_service_client.close()

次に、Blobのアップロードやダウンロード、一覧の取得などを行ってみます。

クイック スタート:Python v12 SDK で BLOB を管理する / コンテナーに BLOB をアップロードする

クイック スタート:Python v12 SDK で BLOB を管理する / コンテナー内の BLOB を一覧表示する

クイック スタート:Python v12 SDK で BLOB を管理する / BLOB をダウンロードする

こんな感じで作成。

def test_access_blob():
    blob_service_client: BlobServiceClient = BlobServiceClient.from_connection_string(connection_string)

    container_name: str = 'my-blob-container'

    container_client: ContainerClient = blob_service_client.create_container(container_name)

    try:
        # upload
        blob_client1: BlobClient = container_client.get_blob_client('hello.txt')
        blob_client1.upload_blob(b'Hello Blob Service!!')

        # upload
        blob_client2: BlobClient = container_client.get_blob_client('languages.txt')
        blob_client2.upload_blob(b'Python, Java, JavaScript')

        # list
        list_blobs: ItemPaged = container_client.list_blobs()

        blobs: list = []

        for blob in list_blobs:
            blob: BlobProperties = blob
            blobs.append(blob.name)

        assert blobs == ['hello.txt', 'languages.txt']

        # download
        assert blob_client1.download_blob().readall().decode('utf8') == 'Hello Blob Service!!'
        assert blob_client2.download_blob().readall().decode('utf8') == 'Python, Java, JavaScript'

        blob_client1.close()
        blob_client2.close()
    finally:
        container_client.delete_container()
        container_client.close()
        blob_service_client.close()

アップロード。バイナリでアップロードするようです。

        # upload
        blob_client1: BlobClient = container_client.get_blob_client('hello.txt')
        blob_client1.upload_blob(b'Hello Blob Service!!')

        # upload
        blob_client2: BlobClient = container_client.get_blob_client('languages.txt')
        blob_client2.upload_blob(b'Python, Java, JavaScript')

一覧の取得。

        # list
        list_blobs: ItemPaged = container_client.list_blobs()

        blobs: list = []

        for blob in list_blobs:
            blob: BlobProperties = blob
            blobs.append(blob.name)

        assert blobs == ['hello.txt', 'languages.txt']

ダウンロード。

        # download
        assert blob_client1.download_blob().readall().decode('utf8') == 'Hello Blob Service!!'
        assert blob_client2.download_blob().readall().decode('utf8') == 'Python, Java, JavaScript'

ところで、このテストコードは先ほどのAzuriteの起動方法だと、テスト実行時に失敗します。

           azure.core.exceptions.HttpResponseError: x-ms-encryption-algorithm header or parameter is not supported in Azurite strict mode. Switch to loose model by Azurite command line parameter "--loose" or Visual Studio Code configuration "Loose". Please vote your wanted features to https://github.com/azure/azurite/issues

具体的には、アップロードで失敗します。Azuriteが認識できないHTTPヘッダーまたはパラメーターがあるようです。

これを回避するには、メッセージに書かれている通りAzuriteの起動オプションに--looseまたは-Lを指定します。

    -L, --loose               Optional. Enable loose mode which ignores unsupported headers and parameter

というわけで、-Lオプションを付けてAzuriteを再起動しておきます。

$ npx azurite -l data -L

こんなところでしょうか。

最後に、作成したテストコード全体を載せておきます。

tests/test_azurite_access.py

from azure.core.paging import ItemPaged
from azure.storage.blob import BlobServiceClient, BlobClient, ContainerClient, BlobProperties

connection_string: str = 'DefaultEndpointsProtocol=http;AccountName=devstoreaccount1;AccountKey=Eby8vdM02xNOcqFlqUwJPLlmEtlCDXJ1OUzFT50uSRZ6IFsuFq2UVErCz4I6tq/K1SZFPTOtr/KBHBeksoGMGw==;BlobEndpoint=http://127.0.0.1:10000/devstoreaccount1;QueueEndpoint=http://127.0.0.1:10001/devstoreaccount1;'


def test_connect_container():
    blob_service_client: BlobServiceClient = BlobServiceClient.from_connection_string(connection_string)

    container_name: str = 'my-blob-container'

    container_client: ContainerClient = blob_service_client.create_container(container_name)

    try:
        list_blobs: ItemPaged = container_client.list_blobs()

        blobs: list = []

        for blob in list_blobs:
            blob: BlobProperties = blob
            blobs.append(blob.name)

        assert len(blobs) == 0
    finally:
        container_client.delete_container()
        container_client.close()
        blob_service_client.close()


def test_access_blob():
    blob_service_client: BlobServiceClient = BlobServiceClient.from_connection_string(connection_string)

    container_name: str = 'my-blob-container'

    container_client: ContainerClient = blob_service_client.create_container(container_name)

    try:
        # upload
        blob_client1: BlobClient = container_client.get_blob_client('hello.txt')
        blob_client1.upload_blob(b'Hello Blob Service!!')

        # upload
        blob_client2: BlobClient = container_client.get_blob_client('languages.txt')
        blob_client2.upload_blob(b'Python, Java, JavaScript')

        # list
        list_blobs: ItemPaged = container_client.list_blobs()

        blobs: list = []

        for blob in list_blobs:
            blob: BlobProperties = blob
            blobs.append(blob.name)

        assert blobs == ['hello.txt', 'languages.txt']

        # download
        assert blob_client1.download_blob().readall().decode('utf8') == 'Hello Blob Service!!'
        assert blob_client2.download_blob().readall().decode('utf8') == 'Python, Java, JavaScript'

        blob_client1.close()
        blob_client2.close()
    finally:
        container_client.delete_container()
        container_client.close()
        blob_service_client.close()