概要

知らないうちに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月正式リリースだったらしい、アンテナゆるゆるでした

参考

MSLearn - Azure Functions の Python 開発者向けガイド