Azure Blob Storage を扱う Python アプリの開発環境を Docker で作る

タイトルの通り、Azure Blob Storage を扱う Python アプリの開発環境を Docker で作りたかったのでメモ。

Azurite

開発環境から本物の Blob Storage を触るのはダルいので、ローカルでそれっぽく動くやつないかググったら Microsoft 公式から Azurite というツールが公開されていた。さすが。

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

Docker イメージを動かす

公式 Docker イメージもあったので Docker Compose で動かしてみる。

compose.yml

services:

  my_python_app:
    (略)

  azurite:
    image: mcr.microsoft.com/azure-storage/azurite
    volumes:
      - azurite:/data

volumes:
  azurite:

今回は my_python_app サービスから触れればよく「特にポート公開とかはしなくていいや」となったのでボリュームを /data にマウントするだけで済んだ。

中ではポート 10000 で Blob Service が listen しているらしい。

Blob Service に接続する

Azurite を動かすと「ローカルに Azure ストレージアカウントがひとつ作られている」みたいな感じになるっぽい。

ストレージアカウントの接続情報は公式ドキュメントに書いてあった。

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

今回は Python から Blob Service を利用するので、こんな感じになった。 (azure-storage-blob パッケージのインストールが必要)

from azure.storage.blob import BlobServiceClient

def get_blob_service_client() -> BlobServiceClient:
    # TODO: 本番環境の場合は本番環境用の BlobServiceClient を返すようにする
    azurite_conn_str = "DefaultEndpointsProtocol=http;AccountName=devstoreaccount1;AccountKey=Eby8vdM02xNOcqFlqUwJPLlmEtlCDXJ1OUzFT50uSRZ6IFsuFq2UVErCz4I6tq/K1SZFPTOtr/KBHBeksoGMGw==;BlobEndpoint=http://azurite:10000/devstoreaccount1;"
    return BlobServiceClient.from_connection_string(azurite_conn_str)


blob_service_client = get_blob_service_client()

azurite_conn_str は公式ドキュメントからコピペしてきて 127.0.0.1 の部分を azurite に変えただけ。

Blob Storage を操作する

公式ドキュメントを見ながら操作していく。
Amazon S3 とノリは同じだったのでだいぶわかりやすかった。

Azure Storage SDK for Python | Microsoft Learn

blob_service.list_containers()  #=> []

まだなんも作ってないので空。それはそう。

blob_container = blob_service.create_container("foo")

Blob コンテナを作ってみた。

blob_container = blob_service.get_container_client("foo")

作成済みの Blob コンテナのクライアントを取得するときはこう。

blob_container.list_blobs()  #=> []

コンテナから Blob 一覧を取得するときはこう。

データを Blob にアップロードしてみる。

data = b"FOOOOOO!!!"
blob_container.upload_blob("foo.txt", data)

data には bytes や str だけでなく Iterable や IO も指定できるらしいので、普通に何でもアップロードできそう。

ファイル一覧を取得してみる。

for blob in blob_container.list_blobs():
    print(blob["name"])
#=> "foo.txt"

ファイルをダウンロードしてローカルに書き出してみる。

stream = blob.download_blob()
with open("foo.txt", "wb") as f:
    f.write(stream.readall())

「せやな」という感じで普通に操作ができた。

詳しいことは公式ドキュメントを参照： Azure Storage SDK for Python | Microsoft Learn