要怎麼在本機環境使用 VS Code 開發 Azure Function,這兩份官方文件已經說明了大概的流程,也很值得參考:
- Develop Azure Functions by using Visual Studio Code
- Quickstart: Create a function in Azure with Python using Visual Studio Code
但是看完之後,實際開發過程還是會遇到一些問題,本文特此留個紀錄。以下說明如何在本機環境中(不需要創建 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