6
6

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

@takashiuesaka(貴志 上坂)
in

Azure Functions を Python で初めて実装する人へ

6
Last updated at Posted at 2025-08-31

はじめに

Azure Functions を Python で初めて実装するお客様へのご案内のために調べたことをメモったものになります。もしかしたらどなたかのお役に立つかも、ということで公開しておきます。内容はあくまでも個人的なメモであり、会社としてのものでは無いことをご承知おきください。

環境

  • Azure Functions Core Tools: 4.2.1
  • Python: 3.11
  • VS Code
  • Azure Functions 動作は Flex Consumptionでホスティング

Blueprint

Azure Functions にはプログラミングモデル V1 と V2 があるのだが、V2 ではデコレーター（属性・アノテーション）でトリガー/アクションを定義できる形式なったことで他の言語と同様とても実装しやすくなっている。

V2 ではルート直下のfunction_app.pyファイルに Trigger をハンドリングする関数(def)を定義していくのだが、複数の Trigger を実装していくとfunction_app.pyが肥大化していってしまう。

FastAPI や Flask 使用可能な Blueprint というモジュール化機能を V2 では使用できるので、これを使って管理するのが良いだろう。

モジュール化といいつつも Endpoint を別ファイルにて実装できるので、Endpoints フォルダ、または Features フォルダを作ってその中にファイルを作成していく形になるだろう。

Application Insights への出力

セットアップが完了していれば、コードの中では logger を使用して出力すれば良いだけ。

from logging import getLogger
logger = getLogger(__name__)

def genericcompletion(req: func.HttpRequest, response: str) -> func.HttpResponse:
    logger.info('Python HTTP trigger function processed a request.')

セットアップさえ終わっていれば、エラーが発生しても自動的にApplication Insightsにログは飛ぶようになっている。

これは Managed Id の権限設定がされていないことによるエラーの例。
appinsights_error1.jpg

例外詳細がわかるので、問題発生時には役に立つ。
appinsights_error2.jpg

閉域化などセキュリティ対策

Private Endpoint を適用することで Public からのリクエストを受け付けなくすることができる。

脅威検知は

  • まずは Microsoft Defender for Cloud の採用
  • 検知精度を上げるなら Microsoft Sentinel の採用

これがAzure上でPaaSを活用する上での基本だが、Azure Functionsで Microsoft Defender for Cloudによる驚異検知をしたい場合、ホスティングプランとして Flex Consumptin を選択できない。

非同期処理

下に書いておいた同時実行制御は1つのインスタンス内のプロセス数の話でAzure Functionsとしてのセットアップが必要なテーマ。一方、非同期処理はスレッドの話で実装にて対応する。

例えば、外部へHTTP Requestを投げるなら、aiohttpパッケージを使って非同期で処理を実装することでスループット向上が見込める。

import azure.functions as func
import logging
import aiohttp

async_test_bp = func.Blueprint()

@async_test_bp.route(route="async_test")
async def async_test(req: func.HttpRequest) -> func.HttpResponse:
    logging.info('Python HTTP trigger function processed a request.')

    async with aiohttp.ClientSession() as client:
        async with client.get("https://www.msn.com/ja-jp") as response:
            return func.HttpResponse(await response.text())

    return func.HttpResponse(body='NotFound', status_code=404)

このように async def でメソッドを定義し、処理中でasyncをつけて非同期処理を実装する。

ちなみにasync with 〜〜はスコープを抜けると自動的に使ったリソースをお片付けしてくれる便利な実装。C#でいうところのusingによるIDisposable自動起動と一緒。__enter__と__exit__を実装しているクラスであれば、この with を使える。

各種トリガー・バインド

実際に試してみるとハマるところがちょいちょいあったり、意味が分かりにくいオプションがあったりするのでメモっておく。

Timer Trigger

実装例

import logging
import azure.functions as func

app = func.FunctionApp()

@app.function_name(name="mytimer")
@app.timer_trigger(
    schedule="*/10 * * * * *",
    arg_name="myTimer",
    run_on_startup=True,
    use_monitor=False,
)
def mytimer_trigger(myTimer: func.TimerRequest) -> None:
    if myTimer.past_due:
        logging.info('The timer is past due!')

    logging.info('The timer is triggerd.')

Timer起動の指定は CRON 式が元々のやり方。後から TimeSpan 式が追加されている。
TimeSpan 式は制限があるので注意。

TimeSpan は、App Service プランで実行している関数アプリに対してのみ使うことができます。

use_monitor は次回起動時間をBlobに書き込むことで、もしインスタンスが再起動しても次回起動に影響がないようにする仕組みらしい。ただし、この例のように1分以内に複数回起動する場合は既定でfalseになるそうだ。いちいちBlobに書くと遅くなるからだろう。

引数で受け取るTimerRequestオブジェクトはこの例の通り past_due プロパティしか無いみたい。

Timer Trigger を設定した関数はスケールアウトしても1つのインスタンスでしか稼働しない、とのこと。

注意すべき仕様

  • Timer Trigger を設定した関数でエラーが発生しても再起動はしない。次の時間に起動するだけ。そのため、データを取得・更新する処理を行う場合の冪等性の担保がすごく重要になる。

  • Timer Trigger を指定した関数を含むインスタンスがスケールアウトしても、実際に稼働する Timer Trigger は1つだけ。

Timer Triggerの詳細はこちらの投稿が詳しい。
https://knstkny.hatenablog.com/entry/azure_function_timer_trigger

CosmosDB 出力バインド

実装例

import logging
import azure.functions as func

app = func.FunctionApp()

@app.route()
@app.cosmos_db_output(arg_name="documents", 
                      database_name="%DB_NAME%",
                      container_name="%COLLECTION_NAME%"
                      create_if_not_exists=True,
                      connection="CosmosDB",
                      partition_key="/id",
)
def main(req: func.HttpRequest, documents: func.Out[func.Document]) -> func.HttpResponse:
    request_body = req.get_body()
    documents.set(func.Document.from_json(request_body))
    return 'OK'

出力バインドのデコレータ（属性・アノテーション）用のメソッドは cosmos_db_output と cosmos_db_output_V3 の2つがある。この例の cosmos_db_output が新しい実装方式で、引数名が V3 とは異なる。困ったことに、Azure Functions Core Toolsで CosmosDB 出力バインドを追加すると、V3の方で作成されるので変更を忘れないように。

database_name, container_name の値である %DB_NAME%, %COLLECTION_NAME% は環境変数から値を取ってくることを意味している。また、 connection の値 CosmosDB は %で囲われていないがこれも環境変数の KEY 値。なので、ローカル実行の場合は次のように追加設定をする。

local.settings.json
{
  "IsEncrypted": false,
  "Values": {
    "AzureWebJobsStorage": "",
    "FUNCTIONS_WORKER_RUNTIME": "python",
+    "CosmosDB": "AccountEndpoint=https://<YOUR_COSMOSDB_NAME>.documents.azure.com:443/;AccountKey=fOcLPpBS6NtI・・・==;",
+    "COSMOS_DATABASE": "MyDB",
+    "COSMOS_CONTAINER": "MyContainer1"
  }
}

当たり前だが、Azure Functionsの方でも同様に環境変数設定をする。
env_setting.jpg

起動間隔がどれぐらい短くても動作するか。限りなく短くしようとしても1秒ごとに起動するように設定するのが限度。ただし、前の処理が1秒以上かかった場合、次の処理は即時実行されることになる。

Azure OpenAI テキスト補完入力バインド

2025/8時点でまだプレビュー機能のため、MS Docsの設定だけでは動かない。まずは host.json を次のように修正する。

host.json
  "extensionBundle": {
-    "id": "Microsoft.Azure.Functions.ExtensionBundle",
+    "id": "Microsoft.Azure.Functions.ExtensionBundle.Preview",
    "version": "[4.*, 5.0.0)"
  }

実装例

import logging
import azure.functions as func

app = func.FunctionApp()

@app.route(route="genericcompletion", methods=["POST"])
@app.text_completion_input(
    arg_name="response",
    prompt="{prompt}",
    chat_model="gpt-4.1",
    ai_connection_name="myAzureOpenAI",
)
def genericcompletion(req: func.HttpRequest, response: str) -> func.HttpResponse:
    logging.info('Python HTTP trigger function processed a request.')

    response_json = json.loads(response)
    return func.HttpResponse(response_json["content"], status_code=200)

外部から入力された値がprompt="{prompt}"の{prompt}に入ってくる。それをAOAIに投げてテキスト補完した結果をresponse: str で受け取っている。つまり、AOAIに投げる箇所を全てライブラリに任せている。

promptには次のようにしてシステムプロンプトを入れ込むこともできる。

prompt="以下の内容について武士のような口調で応答してください。 ###{prompt}",

実験するとこんな感じ。

$ curl -X POST http://localhost:7071/api/genericcompletion -H "Content-Type: application/json" -d '{"prompt":"君をトレーニングしたデータはちょっと古いの？"}'                                                          
拙者に与えられし知識は、令和六年六月までのものにござる。その後の世の移ろいについては、未だ知らぬことも多きゆえ、何卒ご容赦くだされ。新しき事象については、拙者の力及ばぬこともあるやもしれぬ。

text_completion_inputメソッドには model 引数があるがそれは無視して chat_model に対して文字列でデプロイ済みのモデル名をセットする。
また、ai_connection_nameはオプションだが複数のモデルや接続先を使うときのためには必要な設定。AOAIに接続するときの設定をするときの環境変数のKEYのPREFIXを指定している。この例だと値として myAzureOpenAI をセットしたので、次の内容でセットアップすることになる。

local.settings.json
{
  "IsEncrypted": false,
  "Values": {
    "AzureWebJobsStorage": "",
    "FUNCTIONS_WORKER_RUNTIME": "python",
+    "myAzureOpenAI__endpoint": "https://<YOUR_OPENAI_ENDPOINT>.openai.azure.com/",
+    "myAzureOpenAI__key": "・・・"
  }
}

Azure Functionsの環境変数にも次のように設定する。
env_openai1.jpg

Managed ID での接続にする

Keyを使うのはセキュリティ上あまりよろしく無いので、Managed IDでの接続にしたい。
まず Azure Functions の Managed Id を有効化する。
func-managedId.jpg

次に使用しているAOAI、またはAI FoundryリソースをAzure Portalで開いて、Managed Idを有効化しておく。AI Foundry Portalからでは設定できないので注意。

AI Foundry
rbac_ai_foundry.jpg

Azure OpenAI
rbac_openai.jpg

設定するロールは Cognitive Services OpenAI ユーザー ロール。

マネージド ID を使用する場合は、必ず Cognitive Services OpenAI ユーザー ロールに ID を追加してください。

cognitive_openai_user_role.jpg

次のページの一番下に Cognitive Services OpenAI ユーザー ロールを付与することが明記されている。

Azure Functions上の環境変数設定を次のようにする。
<PREFIX>__credential の値は managedidentity で固定。
env_openai2.jpg

Azure Portal -> 設定 -> Azure OpenAI

Azure Portal で Azure Functions を開くと左サイドバーメニューの設定の中に Azure OpenAIという項目がある。既定ではこのように Not connected と表示される。

aoai_portal1.jpg

これが接続状態である場合、というのは AppSettings の環境変数で次の2つの Key が設定されているかどうかを示しているだけに過ぎない。

  • AZURE_OPENAI_ENDPOINT
  • AZURE_OPENAI_KEY

aoai_appsettings.jpg

すると、接続済みとしてセットアップした Endpoint を使用して該当の AOAI リソースへのリンクが表示される。
aoai_portal2.jpg

AOAIを使う場合にこのセットアップが必須なわけではない。すでに示したAOAIの入力バインドで使用するKEY名とも異なる。SDK を使って AOAI と連携する場合にこの KEY 名を使うように実装しておけば、少しだけわかりやすいというものでしかない。

Service Bus トリガー

実装そのものに難しい点は無い。Functions側でのセットアップとして、下に記載した動的な同時実行制御を有効にしておくことで、１インスタンス内で適切にプロセス数を増減してくれるので、有効にしておくことを推奨する。

実装例

import azure.functions as func
import logging

app = func.FunctionApp()

@app.service_bus_queue_trigger(
        arg_name="azservicebus",
        queue_name="%QueueName%",
        connection="msue_SERVICEBUS") 
def servicebus_queue_trigger(azservicebus: func.ServiceBusMessage):
    message = azservicebus.get_body().decode('utf-8')
    logging.info('Python ServiceBus Queue trigger processed a message: %s', message)

    if message.startswith("テスト"):
        raise ValueError("This is a test error triggered by message content.")
local.settings.json
{
  "IsEncrypted": false,
  "Values": {
    "AzureWebJobsStorage": "",
    "FUNCTIONS_WORKER_RUNTIME": "python",
    "msue_SERVICEBUS": "Endpoint=sb://<YOUR_ENDPOINT_NAME>.servicebus.windows.net/;SharedAccessKeyName=・・・",
    "QueueName": "myqueue1"
  }
}

Service Bus からのメッセージ読み取りはPeek-Lockなので読み取った段階ではメッセージはService Bus側に残っているが他からは読み取れない状態になる。このメッセージは関数が正常終了するとメッセージがCompleteになり、異常終了すると Abondon となる。つまり Dead-letter Queueにメッセージが移動する。

うまく接続できない時は、ほぼ間違いなく Azure Service Bus 側のローカル認証(SASキー認証)が無効になっている。有効にする場合、Azure Portal で Azure Service Busの概要ページを開いて設定する。

sb_local_auth.jpg

ローカル認証(SASキー認証)は無効のままの方がセキュリティ的には望ましいが、まず接続がうまくいくかを確認するために有効にし、うまくいったことを確認してから無効にして Managed ID での接続に切り替えることをお勧めする。

Managed ID での接続にする

Azure Functions から Service Bus への接続を Managed IDにすることができる。Azure Functions の Managed ID に対し、Azure Service Bus データ受信者 ロールだけを付与する。（権限付与は必要最低限という原則に基づく）

ローカル実行時にも Managed ID で接続するなら自分のユーザープリンシパルに対してAzure Service Bus データ受信者 ロールを付与しておく。

Azure Functions 側で環境変数に KEY <CONNECTION_NAME_PREFIX>__fullyQualifiedNamespaceの設定が必要。<CONNECTION_NAME_PREFIX>は Trigger 設定の connection に設定した値のこと。上の例で行くと、msue_SERVICEBUSのことになので、

msue_SERVICEBUS__fullyQualifiedNamespace が KEY 名である。

また、値は<Service Bus リソース名>.servicebus.windows.netをセットする。これを Azure Functions の環境変数にセットする。

func_env.jpg

ローカルでも自分のユーザープリンシパルで動かすなら、次のように local.settings.json にセットする。

local.settings.json
{
  "IsEncrypted": false,
  "Values": {
    "AzureWebJobsStorage": "",
    "FUNCTIONS_WORKER_RUNTIME": "python",
+    "<CONNECTION_NAME_PREFIX>__fullyQualifiedNamespace": "<YOUR_SERVICEBUS_NAME>.servicebus.windows.net"
  }
}

同時実行制御 - Concurrency

1つのインスタンスの中で稼働するプロセス数をどう制御するか、というテーマ。ここは抑えておかないとスループットに多大な影響がある。

静的な同時実行制御と動的な同時実行制御の2種類あることをまず抑える。
動的になんでもやってくれりゃ便利なのだが、全ての Trigger には対応していない。

HTTP Trigger

HTTP Trigger は静的な同時実行制御。つまり Concurrency 数を自分で設定しなければいけない。

HTTP Trigger に対する既定の同時実行数は Python の場合 1 になっている。

どれぐらいの数が適切なのかは Performace Optimaizer を活用して決めるのが良い。良いのだが、負荷をかけて稼働した分は有料になってしまうので、気軽にはやらないこと。

Azure Load Testing を使用して、Azure Functions をパフォーマンスとコストに合わせて最適化する

様々な Concurrency数を試した上で、ベストな数を提案してくれる。

Queue, Blob, ServiceBus Trigger

この3つについては動的な同時実行制御がサポートされている。つまり、自動的に制御してくれる。

既定では、動的な同時実行制御は無効になっているので、host.jsonをいじって有効にする。

host.json
{ 
    "version": "2.0", 
    "concurrency": { 
        "dynamicConcurrencyEnabled": true, 
        "snapshotPersistenceEnabled": true 
    } 
}

これ、なぜ既定で無効なのだろう。有効で良いと思う。デメリットがない。

Flex Consumption 使用時の注意

Python に限った話ではないのだが、気になったところをメモっておく。

1. 関数ごとにスケーリングする

これは面白い仕様。複数の関数を1つのAzure Functions にデプロイするのは普通のことだが、それぞれでスケーリングする。ただし、HTTP Trigger, Blob Storage Trigger を指定した関数だけは1つのグループにまとめられてスケールする。

2. Flex 従量課金プランごとに許可されるアプリは 1 つだけ

そもそも従量課金なので App Service プラン使用時のように、複数アプリを動作させて節約する、という方針は検討する必要がない。

3. デプロイスロットがサポートされていない ローリングアップデートで更新できる

今までアプリの更新時には必ず再起動がかかるため、強制的に処理中のセッションも切断されてしまう。いわゆるゼロダウンタイムでのアプリ更新ができなかった。

2025/11のMicrosoft Ignite 2025にてローリングアップデートが発表になった。
http://xn--dhttps-h43e//learn.microsoft.com/ja-jp/azure/azure-functions/flex-consumption-site-updates?tabs=Bicep

これによってゼロダウンタイム更新はサポートされたわけだが、複数インスタンスが稼働している場合は完全に更新が終了するまでは新旧アプリが混在して動作することになるため、下位互換を担保しておく必要があることに注意。

4. スケールアウトの最大値は1000

ただし、既定では最大値は100になっているので設定しないと1000まではスケールアウトしない。

最大値は Azure Portal で設定できる。

max_scaling.jpg

しかし、1000までスケールアウトする前にリージョン・サブスクリプションのメモリクォータという Flex Consumption 独自の制限に引っかかってしまう可能性がある。

現在、特定のサブスクリプション内の各リージョンには、Flex 従量課金プランで実行されているすべてのアプリ インスタンスに対して、 512,000 MB の既定のメモリ制限クォータがあります。

Azure Functions 1つのインスタンスサイズは512MB, 2048MB, 4096MBの3種類から選ぶ。つまり512MBのアプリなら1000までスケールできるが、ややこしいのはメモリの計算は1つの Azure Functionsについてだけ行うのではない、という点。
「全てのアプリ インスタンス」とは別のAzure Functionsであっても同じリージョン・サブスクリプションであれば計算対象となる。

このメモリ制限はサポートチケットで引き上げられる。どこまで引き上がるかは知らない。

5. KeyVault 参照する場合は KeyVault が Public でなければならない

Private 化されている場合、VNET統合していたとしても参照できない。ぐぬぬぬ

なぜ Flex Consumptionを選択するべきか

  • 最大インスタンス数が 1000。他のSKUだとWindows:100, Linux:20。ただし、Flex ConsumptionではLinuxしかない（今時はこれが問題になることはない）
  • 既定のタイムアウトが30分と長い。Premium、AppServiceと同じ。従量課金だと5分と短い（HTTP Triggerはどう設定しても230秒。 これは、Azure Load Balancer の既定のアイドル タイムアウトのため）
  • 常時稼働インスタンスを設定できる。MS Docsには Azure Portalでは設定できない、と記載されているが、設定できる。「スケールとコンカレンシー」メニューを開くと一番下に設定箇所がある
  • VNET統合できる

唯一の弱点はコンテナがホスティングできないことぐらいか。コンテナ化した Functions をデプロしたいなら Azure Container Appsを使うことになる。

プランの比較はこちらを参照すること。
Azure Functions のホスティング オプション

6
6
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
Sign upLogin
6
6

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?