Azure Python

在本機環境使用 VS Code 及 Python 開發 Azure Function 及讀寫 Storage Blob

要怎麼在本機環境使用 VS Code 開發 Azure Function,這兩份官方文件已經說明了大概的流程,也很值得參考:

但是看完之後,實際開發過程還是會遇到一些問題,本文特此留個紀錄。以下說明如何在本機環境中(不需要創建 Azure 資源,連登入 Azure 帳號都不用)使用 VS Code 開發 Azure Function,並且使用 Input Binding 及 BlobClient 讀寫(本機模擬的) Storage Blob

本機環境設定

首先,本機環境需要安裝:

  • Python. 請注意,在截稿前 Azure Function Runtime 只支援到 Python 3.9 (3.10 還在 preview),所以 Python 版本必須是 3.9 或以下。最新的 Python 版本支援資訊可看此
  • VS Code 以及這些 Extensions: Python, Azure Functions (或 Azure Tools), Azurite
  • Azure Functions Core Tools
  • Azure Storage Explorer

創建 Function 及 TimerTrigger Template

接著就可以透過 Azure Functions Extension 在 VS Code 內創建一個 Function. 網路上的範例多為 HTTP Trigger, 但我工作上需要一個 Timer Trigger, 本文也以此為範例

依照提示一步步完成創建之後,就可以 F5 (or Ctrl + F5) 試著執行 Function, 並且右鍵啟動 TimerTrigger

過程中如果遇到 Failed to verify “AzureWebJobsStorage” connection 訊息,就是還沒有啟動 Azurite 的 Blob Service,可以透過上方命令列或是右下角 [Azurite Blob Service] 啟動

另外如果看到 Invalid message: “noDebug” is not supported for “attach” 訊息,這似乎是 Azure Functions Core Tools 在 VS Code 上引起的 bug,並不影響執行。

透過 Input Binding 讀取 Storage Blob

參考官方文件,首先在 function.json中增加 string type 的 binding(此處我是要讀取一個文字檔:在 sample-container 內的 sample.txt)

{
    "name": "inputtxt",
    "type": "blob",
    "dataType": "string",
    "path": "sample-container/sample.txt",
    "connection": "MyConnectionString",
    "direction": "in"
}

此處的 connection 有兩個選擇,可以是 "connection": "AzureWebJobsStorage",或是自定義的 "connection": "MyConnectionString",並且在 local.settings.json 中新增 "MyConnectionString": "UseDevelopmentStorage=true" 在 Values 內

"Values": {
    ...
    "MyConnectionString": "UseDevelopmentStorage=true"
}

本文採用第二個作法,因為稍後寫入 Blob 時還需要回頭修改 MyConnectionString

要怎麼在本機創建一個 Blob container 並上傳檔案?開啟 Azure Storage Explorer,就可以看到本機模擬的 Storage Account 及新增 container(記得先啟動 VS Code 裡面的 Azurite Blob Service)

接著修改 TimerTrigger 的__init__.py

def main(mytimer: func.TimerRequest, inputtxt: str) -> None:
    ...
    logging.info(str(inputtxt))

重新啟動 TimerTrigger 看文字檔是否正確讀入 (line 1 and line 2)

透過 BlobClient 上傳 Storage Blob

為什麼不用 Output Binding 上傳/更新 Blob?因為我還沒有找到使用 Output Binding 要怎麼在程式碼中指定 Blob 的檔名(以下範例的檔名是 sample-upload.txt,但實際上我需要根據今天的日期決定檔名)。

要使用 BlobClient,首先在 requirements.txt 中加入 azure-storage-blob。Function 啟動之前會自動安裝檔案內列出的套件。接著修改 __init__.py

...
from azure.storage.blob import BlobClient
def main(mytimer: func.TimerRequest, inputtxt: str) -> None:
    ...
    blob = BlobClient.from_connection_string(
                conn_str=os.environ["MyConnectionString"], 
				container_name="sample-container", 
                blob_name="sample-upload.txt")
    data = b"Sample data for blob"
    blob.upload_blob(data, overwrite=True)

此處透過 os.environ["MyConnectionString"] 讀取 MyConnectionString 的值。要注意的是須把 MyConnectionString 的值從 UseDevelopmentStorage=true 改成:

DefaultEndpointsProtocol=http;AccountName=devstoreaccount1;AccountKey=Eby8vdM02xNOcqFlqUwJPLlmEtlCDXJ1OUzFT50uSRZ6IFsuFq2UVErCz4I6tq/K1SZFPTOtr/KBHBeksoGMGw==;BlobEndpoint=http://127.0.0.1:10000/devstoreaccount1;

相關說明請看此。azure-storage-blob library 的其他用法可參考此處

再度執行 TimerTrigger 後,可以從 Azure Storage Explorer 看到檔案已經上傳

完整相關檔案如下:

function.json

{
  "scriptFile": "__init__.py",
  "bindings": [
    {
      "name": "mytimer",
      "type": "timerTrigger",
      "direction": "in",
      "schedule": "* * 8 1 * *"
    }, 
    {
      "name": "inputtxt",
      "type": "blob",
      "dataType": "string",
      "path": "sample-container/sample.txt",
      "connection": "MyConnectionString",
      "direction": "in"
    }
  ]
}

__init__.py

import datetime
import logging
import os
from azure.storage.blob import BlobClient
import azure.functions as func
def main(mytimer: func.TimerRequest, inputtxt: str) -> None:
    utc_timestamp = datetime.datetime.utcnow().replace(
        tzinfo=datetime.timezone.utc).isoformat()
    if mytimer.past_due:
        logging.info('The timer is past due!')
    logging.info('Python timer trigger function ran at %s', utc_timestamp)
    logging.info(str(inputtxt))
    blob = BlobClient.from_connection_string(
               conn_str=os.environ["MyConnectionString"], 
               container_name="sample-container", 
               blob_name="sample-upload.txt")
    data = b"Sample data for blob"
    blob.upload_blob(data, overwrite=True)

local.settings.json

{
  "IsEncrypted": false,
  "Values": {
    "AzureWebJobsStorage": "UseDevelopmentStorage=true",
    "FUNCTIONS_WORKER_RUNTIME": "python",
    "MyConnectionString": "DefaultEndpointsProtocol=http;AccountName=devstoreaccount1;AccountKey=Eby8vdM02xNOcqFlqUwJPLlmEtlCDXJ1OUzFT50uSRZ6IFsuFq2UVErCz4I6tq/K1SZFPTOtr/KBHBeksoGMGw==;BlobEndpoint=http://127.0.0.1:10000/devstoreaccount1;"
  }
}

requirements.txt

azure-functions
azure-storage-blob