FunctionsからBlob Storageに保管しているファイルを取得する備忘録

環境

Azure Functions

こちらの記事を参照して用意する。VSCodeでの開発環境の構築も含んでます。

Blob Storage

ストレージアカウントを新規に作成する

詳細タブの匿名アクセスは有効にしない。ほかはデフォルトのまま作成までやる

ストレージのコンテナにファイルをアップロードする

作成したリソースに移動し、左のメニューのコンテナから新しく「faq」という名前のコンテナを作成する

作成したコンテナでファイルをアップロードする

今回はFAQのテキストファイルをアップロードしていて、中身はこんな感じのJSONです

{
  "category": "質問",
  "title": "アプリの基本機能について",
  "detail": "初めてアプリを使うのですが、基本的な使い方や機能を教えてください。",
  "response": "初めてのご利用ありがとうございます。アプリの基本機能としては、運動管理、食事管理、目標設定、進捗確認があります。アプリ内のヘルプセクションで詳しい使い方をご覧いただけます。"
}

このファイルは、フィットネス系のアプリをリリースしたという設定で、ユーザからの問い合わせについて想定問答集作ってとChatGPTに適当に吐き出させたファイルです。

関数を作る

HTTPのトリガーで、先ほど作成したコンテナからファイルのリストを取得して、先頭データのファイルの中身を応答文として返します。

function_app.py

import azure.functions as func
import logging
import json
import os
from azure.storage.blob import BlobServiceClient, ContainerClient
from azure.core.exceptions import ResourceNotFoundError

app = func.FunctionApp(http_auth_level=func.AuthLevel.FUNCTION)

#Blobストレージのコンテナからファイルを取得して中身を返す関数
@app.route(route="http_trigger001")
def http_trigger001(req: func.HttpRequest) -> func.HttpResponse:
    logging.info('Python HTTP trigger function processed a request.')
    try:
        # ストレージアカウントへの接続文字列を環境変数から取得して設定
        connection_string = os.getenv("ConnectionString_AzureDataFileStorage")
        if not connection_string:
            raise ValueError("Connection string for 'AzureDataFileStorage' is not set in the environment variables.")
        
        #取得したいデータが格納されたコンテナ名を環境変数から取得して設定
        target_container_name = os.getenv("DataStorage_ContainerName")
        if not target_container_name:
            raise ValueError("Container name 'DataStorage_ContainerName' is not set in the environment variables.")

        # BlobServiceClient を作成
        blob_service_client = BlobServiceClient.from_connection_string(connection_string)
        logging.info('Successfully connected to BlobServiceClient')

        # コンテナクライアントを作成
        container_client = blob_service_client.get_container_client(target_container_name)

        try:
            # コンテナの存在を確認
            container_client.get_container_properties()
            logging.info(f'Container "{target_container_name}" exists.')
        except ResourceNotFoundError:
            logging.error(f'Container "{target_container_name}" does not exist.')
            return func.HttpResponse(
                body=json.dumps({"error": f'Container "{target_container_name}" does not exist.'}),
                status_code=404,
                mimetype="application/json"
            )

        # コンテナ内のすべての BLOB をリストする
        blob_list = container_client.list_blobs()

        # 最初の BLOB を取得
        first_blob = next(blob_list, None)
        if first_blob is None:
            logging.error(f'No blobs found in the container "{target_container_name}".')
            return func.HttpResponse(
                body=json.dumps({"error": f'No blobs found in the container "{target_container_name}".'}),
                status_code=404,
                mimetype="application/json"
            )

        # BLOB クライアントを作成してデータをダウンロード
        blob_client = container_client.get_blob_client(first_blob)
        blob_content = blob_client.download_blob().readall()

        # BLOB の内容を JSON データとしてパース
        try:
            json_data = json.loads(blob_content)
        except json.JSONDecodeError as jde:
            logging.error(f"JSONDecodeError: {str(jde)}")
            return func.HttpResponse(
                body=json.dumps({"error": f"Failed to decode JSON: {str(jde)}"}),
                status_code=500,
                mimetype="application/json"
            )

        # JSON データをレスポンスとして返す
        return func.HttpResponse(
            body=json.dumps(json_data, ensure_ascii=False),
            status_code=200,
            mimetype="application/json"
        )
    except ValueError as ve:
        logging.error(f"ValueError: {str(ve)}")
        return func.HttpResponse(
            body=json.dumps({"error": str(ve)}),
            status_code=500,
            mimetype="application/json"
        )
    except Exception as e:
        logging.error(f"An error occurred: {str(e)}")
        return func.HttpResponse(
            body=json.dumps({"error": str(e)}),
            status_code=500,
            mimetype="application/json"
        )

VSCodeでテストする際は環境変数をlocal.settings.jsonに記載します。
「"Your Storage Connection String"」の部分は自身のストレージアカウントの接続文字列に置き換えてください。（AzureのPortal上で作成したストレージアカウントのリソースを開くと左のメニューに「アクセスキー」という項目あるのでそこで確認できます）
DataStorage_ContainerNameには先ほど作成した「faq」というコンテナ名を入れておきます。

local.settings.json

{
  "IsEncrypted": false,
  "Values": {
    "AzureWebJobsStorage": "",
    "FUNCTIONS_WORKER_RUNTIME": "python",
    "AzureWebJobsFeatureFlags": "EnableWorkerIndexing",
    "ConnectionString_AzureDataFileStorage":"Your Storage Connection String",
    "DataStorage_ContainerName":"faq"
  }
}

Requirements.txtの中身はこれだけ

azure-functions
azure-storage-blob

ローカルで実行

コマンドパレットからエミュレータを起動

Azurite:start

function_app.pyを開いてF5キーで実行する
エラーがなければVSCodeのターミナルにローカルのエンドポイントが出力される

ワークスペース(ローカル)から関数を選択し、「Execute Function Now」をクリックして実行する。
要求本文の引数は今回はなくてもいい（デフォルトのままEnterキー押しても問題ない）。

うまくいけばVSCodeの通知にレスポンスが表示される

Azureにデプロイする

具体的な手順はこちらを参照してください。

デプロイ後の動作確認方法は過去記事でも触れてるので省きますが、
はAzureのポータル上からlocal.settings.jsonに設定した環境変数と同じものを設定することを忘れずに！