概要
知らないうちにPythonのAzure Functionsでmodel V2が正式リリースされていたので
確認がてら以前のV1と比較しつつまとめました。
↑ VSCodeの拡張がプロジェクト作成時に選べるように
環境等
- Windows11
- Python(v3.10.11)
- VSCode, AzureFunctions拡張(v1.13.1)
大きな変更点
- function.jsonが消え、フォルダ単位で関数を管理する必要が無くなった
- バインドは関数のデコレーターが担うように
- blueprintファイル(function_app.py)を使用してのルーティング
- 関数を一括で管理できるように
ファイル構造比較
- 左がV1、右がV2となっている
- V1が基本フォルダ毎に関数を分けているのに対し、V2は関数をモジュール化して
function_app.pyで一元管理している
基本構造
function_app.py
- ファイル名は
function_app.py固定- エントリポイントの為
function_app.py
import azure.functions as func
import logging
app = func.FunctionApp(http_auth_level=func.AuthLevel.ANONYMOUS)
@app.route(route="http_trigger")
def http_trigger(req: func.HttpRequest) -> func.HttpResponse:
logging.info('Python HTTP trigger function processed a request.')
name = req.params.get('name')
if not name:
try:
req_body = req.get_json()
except ValueError:
pass
else:
name = req_body.get('name')
if name:
return func.HttpResponse(f"Hello, {name}. This HTTP triggered function executed successfully.")
else:
return func.HttpResponse(
"This HTTP triggered function executed successfully. Pass a name in the query string or in the request body for a personalized response.",
status_code=200
)
blueprint.py
- ファイル名は任意
- azure.functions.Blueprintクラスのインスタンスに関数を登録しモジュール化している
TimerTrigger.py
import azure.functions as func
import logging
TimerTrigger = func.Blueprint()
@TimerTrigger.timer_trigger(schedule="0 * * * * *", arg_name="myTimer", run_on_startup=True,
use_monitor=False)
def timer_trigger(myTimer: func.TimerRequest) -> None:
if myTimer.past_due:
logging.info('The timer is past due!')
logging.info('Python timer trigger function executed.')
関数のインポート
- azure.functions.FunctionAppインスタンスにインポートしたBlueprintクラスを登録する
function_app.py
import azure.functions as func
from functions.TimerTrigger import TimerTrigger
import logging
app = func.FunctionApp(http_auth_level=func.AuthLevel.ANONYMOUS)
app.register_functions(TimerTrigger)
主なトリガー
HTTP Trigger
- バインド関連が丸っと消えシンプルに
- デコレーターでルーティングをするだけでOK、元の
__init__.pyの実行処理をそのまま移せる
V1
- functions.jsonでin-outのバインディングが必要
__init__.py
def main(req: func.HttpRequest) -> func.HttpResponse:
logging.info('Python HTTP trigger function processed a request.')
name = req.params.get('name')
if not name:
try:
req_body = req.get_json()
except ValueError:
pass
else:
name = req_body.get('name')
if name:
return func.HttpResponse(f"Hello, {name}. This HTTP triggered function executed successfully.")
else:
return func.HttpResponse(
"This HTTP triggered function executed successfully. Pass a name in the query string or in the request body for a personalized response.",
status_code=200
)
functions.json
{
"scriptFile": "__init__.py",
"bindings": [
{
"authLevel": "anonymous",
"type": "httpTrigger",
"direction": "in",
"name": "req",
"methods": [
"get",
"post"
]
},
{
"type": "http",
"direction": "out",
"name": "$return"
}
]
}
V2
- デコレーター
route()によりトリガーバインドを実現
function_app.py
@app.route(route="http_trigger")
def http_trigger(req: func.HttpRequest) -> func.HttpResponse:
logging.info('Python HTTP trigger function processed a request.')
name = req.params.get('name')
if not name:
try:
req_body = req.get_json()
except ValueError:
pass
else:
name = req_body.get('name')
if name:
return func.HttpResponse(f"Hello, {name}. This HTTP triggered function executed successfully.")
else:
return func.HttpResponse(
"This HTTP triggered function executed successfully. Pass a name in the query string or in the request body for a personalized response.",
status_code=200
)
Timer Trigger
- 同様にトリガーバインドがデコレーターで記述できるように
V1
__init__.py
def main(mytimer: func.TimerRequest) -> 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)
functions.json
{
"scriptFile": "__init__.py",
"bindings": [
{
"name": "mytimer",
"type": "timerTrigger",
"direction": "in",
"schedule": "0 */5 * * * *"
}
]
}
V2
- デコレーター
timer_trigger()でバインディング - functions.jsonの値と対応する引数は以下の通り
| functions.json | デコレーター | 概要 |
|---|---|---|
| schedule | schedule | CRON・TimeSpanによる実行間隔 |
| name | arg_name | 関数内のタイマーオブジェクト名(引数名) |
| runOnStartup | run_on_startup | アプリ起動直後の実行の有無 |
| useMonitor | use_monitor | スケジュールの監視 |
TimerTrigger.py
TimerTrigger = func.Blueprint()
@TimerTrigger.timer_trigger(schedule="0 * * * * *", arg_name="myTimer", run_on_startup=True,
use_monitor=False)
def timer_trigger(myTimer: func.TimerRequest) -> None:
if myTimer.past_due:
logging.info('The timer is past due!')
logging.info('Python timer trigger function executed.')
Blob Trigger
V1
- 同様にトリガーバインドがデコレーターで記述できるように
__init__.py
def main(myblob: InputStream):
logging.info(f"Python blob trigger function processed blob \n"
f"Name: {myblob.name}\n"
f"Blob Size: {myblob.length} bytes")
functions.json
{
"scriptFile": "__init__.py",
"bindings": [
{
"name": "myblob",
"type": "blobTrigger",
"direction": "in",
"path": "samples-workitems/{name}",
"connection": "xxx"
}
]
}
V2
- デコレーター
blob_trigger()でバインディング - functions.jsonの値と対応する引数は以下の通り
| functions.json | デコレーター | 概要 |
|---|---|---|
| name | arg_name | 関数内のBlobオブジェクト名(引数名) |
| path | path | 監視対象コンテナー・ファイル |
| connection | connection | ストレージ アカウントの接続文字列 |
BlobTrigger.py
@BlobTrigger.blob_trigger(arg_name="myblob", path="samples-workitems/{name}", connection="xxx")
def blob_trigger(myblob: func.InputStream):
logging.info(f"Python blob trigger function processed blob"
f"Name: {myblob.name}"
f"Blob Size: {myblob.length} bytes")
あとがき
神アプデです。関数の管理が格段に楽になりました。
これまで設定を見るのに一々functions.jsonを開いていたのが不要になり、またプロジェクトルート直下に関数を並べなくて良くなりました。
ちなみにDurableFunctionsが無かったんですがいつ来るので…？
追伸
去年5月正式リリースだったらしい、アンテナゆるゆるでした