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

Azure Functions では、関数は入力を処理して出力を生成する Python スクリプト内のステートレスなメソッドであることが求められます。 既定で、ランタイムでは、このメソッドは main() と呼ばれるグローバル メソッドとして _init_.py ファイル内に実装されると想定されます。 また、代替エントリ ポイントを指定することもできます。

トリガーとバインディングからのデータをメソッド属性を介して関数にバインドするには、function.json ファイル内で定義されている name プロパティを使用します。 たとえば、次の function.json ファイルには、req という名前の HTTP 要求によってトリガーされるシンプルな関数が記述されています。

{
    "scriptFile": "__init__.py",
    "bindings": [
        {
            "authLevel": "function",
            "type": "httpTrigger",
            "direction": "in",
            "name": "req",
            "methods": [
                "get",
                "post"
            ]
        },
        {
            "type": "http",
            "direction": "out",
            "name": "$return"
        }
    ]
}

この定義に基づき、関数コードを含む _init_.py ファイルは次の例のようになります。

def main(req):
    user = req.params.get('user')
    return f'Hello, {user}!'

Python の型の注釈を使用することで、関数内に属性の種類および戻り値の型を明示的に宣言することもできます。 こうすることで、多くの Python コード エディターによって提供される IntelliSense とオートコンプリート機能を使うのに役立ちます。

import azure.functions

def main(req: azure.functions.HttpRequest) -> str:
    user = req.params.get('user')
    return f'Hello, {user}!'

azure.functions.* パッケージに含まれる Python の注釈を使用すると、入力と出力がご利用のメソッドにバインドされます。

Azure Functions では、関数は入力を処理して出力を生成する Python スクリプト内のステートレスなメソッドであることが求められます。 既定で、ランタイムでは、このメソッドはグローバル メソッドとして function_app.py ファイル内に実装されると想定されます。

デコレーター ベースのアプローチでは、トリガーとバインドは関数で宣言および使用できます。 これらは、関数と同じファイル function_app.py で定義されています。 例として、次の function_app.py ファイルは、HTTP 要求によってトリガーされる関数を表します。

@app.function_name(name="HttpTrigger1")
@app.route(route="req")
def main(req):
    user = req.params.get("user")
    return f"Hello, {user}!"

Python の型の注釈を使用することで、関数内に属性の種類および戻り値の型を明示的に宣言することもできます。 こうすることで、多くの Python コード エディターによって提供される IntelliSense とオートコンプリート機能を使うのに役立ちます。

import azure.functions as func

app = func.FunctionApp()

@app.function_name(name="HttpTrigger1")
@app.route(route="req")
def main(req: func.HttpRequest) -> str:
    user = req.params.get("user")
    return f"Hello, {user}!"

v2 モデルに関する既知の制限事項とその回避策については、「Azure Functions での Python エラーのトラブルシューティング」を参照してください。

scriptFile ファイル内で scriptFile プロパティと entryPoint プロパティをオプションで指定することによって、関数の既定の動作を変更できます。 たとえば、以下の function.json では、ご利用の Azure 関数のエントリ ポイントとして、main.py ファイル内の customentry() メソッドを使用するようにランタイムに指示が出されます。

{
  "scriptFile": "main.py",
  "entryPoint": "customentry",
  "bindings": [
      ...
  ]
}

エントリ ポイントは function_app.py ファイル内のみにあります。 ただし、プロジェクト内の関数は、ブループリントを使用するかインポートして、function_app.py で参照できます。

Python 関数プロジェクトの推奨フォルダー構造は、次の例のようになります。

 <project_root>/
 | - .venv/
 | - .vscode/
 | - my_first_function/
 | | - __init__.py
 | | - function.json
 | | - example.py
 | - my_second_function/
 | | - __init__.py
 | | - function.json
 | - shared_code/
 | | - __init__.py
 | | - my_first_helper_function.py
 | | - my_second_helper_function.py
 | - tests/
 | | - test_my_second_function.py
 | - .funcignore
 | - host.json
 | - local.settings.json
 | - requirements.txt
 | - Dockerfile

メイン プロジェクト フォルダー <> には、次のファイルを含めることができます。

• local.settings.json:ローカルで実行するときに、アプリの設定と接続文字列を格納するために使用されます。 このファイルは Azure に公開されません。 詳細については、「local.settings.file」に関するページを参照してください。
• requirements.txt:Azure に公開するときにシステムによってインストールされる Python パッケージの一覧が含まれます。
• host.json: 関数アプリ インスタンス内にあるすべての関数に影響する構成オプションが含まれます。 このファイルは Azure に公開されます。 ローカルで実行する場合は、すべてのオプションがサポートされるわけではありません。 詳細については、「host.json」に関するページを参照してください。
• .vscode/: (省略可能) 格納されている Visual Studio Code 構成が含まれます。 詳細については、Visual Studio Code の設定に関するページを参照してください。
• .venv/ :(省略可能) ローカル開発で使用される Python 仮想環境が含まれます。
• Dockerfile:(省略可能) カスタム コンテナーでプロジェクトを発行するときに使用されます。
• tests/ :(省略可能) 関数アプリのテスト ケースが含まれます。
• .funcignore:(省略可能) Azure に発行しないファイルを宣言します。 通常、このファイルには、エディター設定を無視する場合は .vscode/、ローカルの Python 仮想環境を無視する場合は .venv/、テスト ケースを無視する場合は tests/、ローカル アプリの設定を発行しない場合は local.settings.json が含まれます。

各関数には、独自のコード ファイルとバインディング構成ファイル function.json があります。

Python 関数プロジェクトの推奨フォルダー構造は、次の例のようになります。

 <project_root>/
 | - .venv/
 | - .vscode/
 | - function_app.py
 | - additional_functions.py
 | - tests/
 | | - test_my_function.py
 | - .funcignore
 | - host.json
 | - local.settings.json
 | - requirements.txt
 | - Dockerfile

メイン プロジェクト フォルダー <> には、次のファイルを含めることができます。

• .venv/: (省略可能) ローカル開発に使用される Python 仮想環境が含まれます。
• .vscode/: (省略可能) 格納されている Visual Studio Code 構成が含まれます。 詳細については、Visual Studio Code の設定に関するページを参照してください。
• function_app.py: すべての関数とそれに関連するトリガーとバインドの既定の場所です。
• additional_functions.py: (省略可能) ブループリントを介して function_app.py で参照される関数 (通常は論理グループ化用) を含む他の Python ファイル。
• tests/ :(省略可能) 関数アプリのテスト ケースが含まれます。
• .funcignore:(省略可能) Azure に発行しないファイルを宣言します。 通常、このファイルには、エディター設定を無視する場合は .vscode/、ローカルの Python 仮想環境を無視する場合は .venv/、テスト ケースを無視する場合は tests/、ローカル アプリの設定を発行しない場合は local.settings.json が含まれます。
• host.json: 関数アプリ インスタンス内にあるすべての関数に影響する構成オプションが含まれます。 このファイルは Azure に公開されます。 ローカルで実行する場合は、すべてのオプションがサポートされるわけではありません。 詳細については、「host.json」に関するページを参照してください。
• local.settings.json: ローカルで実行するとき、アプリ設定と接続文字列を格納するために使用されます。 このファイルは Azure に公開されません。 詳細については、「local.settings.file」に関するページを参照してください。
• requirements.txt: Azure に公開するときにシステムによってインストールされる Python パッケージの一覧が含まれます。
• Dockerfile:(省略可能) カスタム コンテナーでプロジェクトを発行するときに使用されます。

Blueprints

Python v2 プログラミング モデルでは、ブループリント の概念が導入されています。 ブループリントは、コア関数アプリケーションの外部で関数を登録するためにインスタンス化された新しいクラスです。 ブループリント インスタンスに登録されている関数は、関数ランタイムによって直接インデックス付けされません。 これらのブループリント関数のインデックスを作成するには、関数アプリはブループリント インスタンスから関数を登録する必要があります。

ブループリントの使用には次のような利点があります。

• 関数アプリをモジュール型コンポーネントに分割して、複数の Python ファイルで関数を定義し、ファイルごとに異なるコンポーネントに分割できます。
• 独自の API をビルドして再利用するための、拡張可能なパブリック関数アプリ インターフェイスを提供します。

次の例は、ブループリントの使用方法を示しています。

まず、http_blueprint.py ファイルで HTTP によってトリガーされる関数が最初に定義され、ブループリント オブジェクトに追加されます。

import logging 

import azure.functions as func 

bp = func.Blueprint() 

@bp.route(route="default_template") 
def default_template(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 " 
            f"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 
        ) 

次に、function_app.py ファイルでブループリント オブジェクトがインポートされ、その関数が関数アプリに登録されます。

import azure.functions as func 
from http_blueprint import bp

app = func.FunctionApp() 

app.register_functions(bp) 

インポートの動作

絶対と相対の各参照の両方を使用して、関数コードにモジュールをインポートすることができます。 次のインポートは、前述のフォルダー構造に基づいて、関数ファイル <project_root>\my_first_function\__init__.py 内から機能します:

from shared_code import my_first_helper_function #(absolute)

import shared_code.my_second_helper_function #(absolute)

from . import example #(relative)

次の __app__ インポートおよび最上位を超える相対インポートは、静的な型チェッカーでサポートされておらず、Python テスト フレームワークでサポートされていないため非推奨です。

from __app__.shared_code import my_first_helper_function #(deprecated __app__ import)

from ..shared_code import my_first_helper_function #(deprecated beyond top-level relative import)

Azure Functions では、入力はトリガー入力とその他の入力の 2 つのカテゴリに分けられます。 function.json ファイルではこれらは同じではありませんが、Python コードでは同じように使用されます。 トリガーの接続文字列またはシークレットと入力ソースは、ローカル実行時には local.settings.json ファイル内の値にマップし、Azure での実行時にはアプリケーションの設定にマップします。

たとえば、次のコードでは 2 つの入力の違いが示されています。

// function.json
{
  "scriptFile": "__init__.py",
  "bindings": [
    {
      "name": "req",
      "direction": "in",
      "type": "httpTrigger",
      "authLevel": "anonymous",
      "route": "items/{id}"
    },
    {
      "name": "obj",
      "direction": "in",
      "type": "blob",
      "path": "samples/{id}",
      "connection": "STORAGE_CONNECTION_STRING"
    }
  ]
}

// local.settings.json
{
  "IsEncrypted": false,
  "Values": {
    "FUNCTIONS_WORKER_RUNTIME": "python",
    "STORAGE_CONNECTION_STRING": "<AZURE_STORAGE_CONNECTION_STRING>",
    "AzureWebJobsStorage": "<azure-storage-connection-string>"
  }
}

# __init__.py
import azure.functions as func
import logging

def main(req: func.HttpRequest, obj: func.InputStream):
    logging.info(f'Python HTTP-triggered function processed: {obj.read()}')

この関数が呼び出されると、HTTP 要求は req として関数に渡されます。 エントリは、ルート URL 内の ID に基づいて Azure Blob Storage アカウントから取得され、関数の本体で obj として使用できるようになります。 ここで、指定されるストレージ アカウントは、CONNECTION_STRING アプリ設定で見つかる接続文字列です。

Azure Functions では、入力はトリガー入力とその他の入力の 2 つのカテゴリに分けられます。 これらは異なるデコレーターを使用して定義されていますが、Python コードでは使用方法は似ています。 トリガーの接続文字列またはシークレットと入力ソースは、ローカル実行時には local.settings.json ファイル内の値にマップし、Azure での実行時にはアプリケーションの設定にマップします。

例として、次のコードは、BLOB ストレージの入力バインドを定義する方法を示しています。

// local.settings.json
{
  "IsEncrypted": false,
  "Values": {
    "FUNCTIONS_WORKER_RUNTIME": "python",
    "STORAGE_CONNECTION_STRING": "<AZURE_STORAGE_CONNECTION_STRING>",
    "AzureWebJobsStorage": "<azure-storage-connection-string>"
  }
}

# function_app.py
import azure.functions as func
import logging

app = func.FunctionApp()

@app.route(route="req")
@app.read_blob(arg_name="obj", path="samples/{id}", 
               connection="STORAGE_CONNECTION_STRING")
def main(req: func.HttpRequest, obj: func.InputStream):
    logging.info(f'Python HTTP-triggered function processed: {obj.read()}')

この関数が呼び出されると、HTTP 要求は req として関数に渡されます。 エントリは、ルート URL 内の ID に基づいて Azure Blob Storage アカウントから取得され、関数の本体で obj として使用できるようになります。 ここで、指定されるストレージ アカウントは、STORAGE_CONNECTION_STRING アプリ設定で見つかる接続文字列です。

Functions では、Azure Blob Storage の Python SDK 型バインドがサポートされています。これにより、基になる BlobClient 型を使用して BLOB データを操作できます。

前提条件

• Azure Functions ランタイム バージョン 4.34 以降のバージョン。
• Python バージョン 3.9 以降のサポートされているバージョン。

Blob Storage 拡張機能の SDK 型バインドを有効にする

1. azurefunctions-extensions-bindings-blob 拡張機能パッケージをプロジェクト内の requirements.txt ファイルに追加します。これには、少なくとも次のパッケージが含まれている必要があります。

   azure-functions
   azurefunctions-extensions-bindings-blob
   

2. 次のコードをプロジェクト内の function_app.py ファイルに追加します。これにより、SDK 型バインドがインポートされます。

   import azurefunctions.extensions.bindings.blob as blob
   

SDK 型バインドの例

この例では、BLOB Storage トリガー (blob_trigger) と HTTP トリガーの入力バインド (blob_input) の両方から BlobClient を取得する方法を示します。

import logging

import azure.functions as func
import azurefunctions.extensions.bindings.blob as blob

app = func.FunctionApp(http_auth_level=func.AuthLevel.ANONYMOUS)

@app.blob_trigger(
    arg_name="client", path="PATH/TO/BLOB", connection="AzureWebJobsStorage"
)
def blob_trigger(client: blob.BlobClient):
    logging.info(
        f"Python blob trigger function processed blob \n"
        f"Properties: {client.get_blob_properties()}\n"
        f"Blob content head: {client.download_blob().read(size=1)}"
    )


@app.route(route="file")
@app.blob_input(
    arg_name="client", path="PATH/TO/BLOB", connection="AzureWebJobsStorage"
)
def blob_input(req: func.HttpRequest, client: blob.BlobClient):
    logging.info(
        f"Python blob input function processed blob \n"
        f"Properties: {client.get_blob_properties()}\n"
        f"Blob content head: {client.download_blob().read(size=1)}"
    )
    return "ok"

Blob Storage の他の SDK 型バインドのサンプルは、Python 拡張機能リポジトリで確認できます。

• ContainerClient 型
• StorageStreamDownloader 型

前提条件

• Azure Functions ランタイム バージョン 4.34.1 以降。
• Python バージョン 3.8 以降のサポートされているバージョン。

HTTP ストリームを有効にする

HTTP ストリームは既定で無効になっています。 アプリケーション設定でこの機能を有効にし、FastAPI パッケージを使用するようにコードを更新する必要があります。 HTTP ストリームを有効にすると、関数アプリでは既定で HTTP ストリーミングが使用されるため、元の HTTP 機能は機能しないことに注意してください。

1. azurefunctions-extensions-http-fastapi 拡張機能パッケージをプロジェクト内の requirements.txt ファイルに追加します。これには、少なくとも次のパッケージが含まれている必要があります。

   azure-functions
   azurefunctions-extensions-http-fastapi
   

2. 次のコードをプロジェクト内の function_app.py ファイルに追加します。これにより、FastAPI 拡張機能がインポートされます。

   from azurefunctions.extensions.http.fastapi import Request, StreamingResponse
   

3. Azure にデプロイするときに、関数アプリに次のアプリケーション設定を追加します。

   "PYTHON_ENABLE_INIT_INDEXING": "1"

   Linux 従量課金プランにデプロイする場合は、以下も追加します

   "PYTHON_ISOLATE_WORKER_DEPENDENCIES": "1"

   ローカルで実行する場合は、これらの同じ設定を local.settings.json プロジェクト ファイルに追加する必要もあります。

HTTP ストリームの例

HTTP ストリーミング機能を有効にすると、HTTP 経由でデータをストリーミングする関数を作成できます。

この例は、HTTP 応答データをストリーミングする HTTP でトリガーされる関数です。 これらの機能を使用して、リアルタイムの視覚化のためにパイプラインを介してイベント データを送信したり、大規模なデータ セットの異常を検出したり、インスタント通知を提供したりするなどのシナリオをサポートできます。

import time

import azure.functions as func
from azurefunctions.extensions.http.fastapi import Request, StreamingResponse

app = func.FunctionApp(http_auth_level=func.AuthLevel.ANONYMOUS)


def generate_sensor_data():
    """Generate real-time sensor data."""
    for i in range(10):
        # Simulate temperature and humidity readings
        temperature = 20 + i
        humidity = 50 + i
        yield f"data: {{'temperature': {temperature}, 'humidity': {humidity}}}\n\n"
        time.sleep(1)


@app.route(route="stream", methods=[func.HttpMethod.GET])
async def stream_sensor_data(req: Request) -> StreamingResponse:
    """Endpoint to stream real-time sensor data."""
    return StreamingResponse(generate_sensor_data(), media_type="text/event-stream")

この例は、クライアントからのストリーミング データをリアルタイムで受信して処理する、HTTP によってトリガーされる関数です。 これは、継続的なデータ ストリームの処理や IoT デバイスからのイベント データの処理などのシナリオに役立つストリーミング アップロード機能を示します。

import azure.functions as func
from azurefunctions.extensions.http.fastapi import JSONResponse, Request

app = func.FunctionApp(http_auth_level=func.AuthLevel.ANONYMOUS)


@app.route(route="streaming_upload", methods=[func.HttpMethod.POST])
async def streaming_upload(req: Request) -> JSONResponse:
    """Handle streaming upload requests."""
    # Process each chunk of data as it arrives
    async for chunk in req.stream():
        process_data_chunk(chunk)

    # Once all data is received, return a JSON response indicating successful processing
    return JSONResponse({"status": "Data uploaded and processed successfully"})


def process_data_chunk(chunk: bytes):
    """Process each data chunk."""
    # Add custom processing logic here
    pass

HTTP ストリームの呼び出し

関数の FastAPI エンドポイントにストリーミング呼び出しを行うには、HTTP クライアント ライブラリを使用する必要があります。 お使いのクライアント ツールまたはブラウザーが、ストリーミングをネイティブにサポートしていないか、データの最初のチャンクのみを返す場合があります。

次のようなクライアント スクリプトを使用して、ストリーミング データを HTTP エンドポイントに送信できます。

import httpx # Be sure to add 'httpx' to 'requirements.txt'
import asyncio

async def stream_generator(file_path):
    chunk_size = 2 * 1024  # Define your own chunk size
    with open(file_path, 'rb') as file:
        while chunk := file.read(chunk_size):
            yield chunk
            print(f"Sent chunk: {len(chunk)} bytes")

async def stream_to_server(url, file_path):
    timeout = httpx.Timeout(60.0, connect=60.0)
    async with httpx.AsyncClient(timeout=timeout) as client:
        response = await client.post(url, content=stream_generator(file_path))
        return response

async def stream_response(response):
    if response.status_code == 200:
        async for chunk in response.aiter_raw():
            print(f"Received chunk: {len(chunk)} bytes")
    else:
        print(f"Error: {response}")

async def main():
    print('helloworld')
    # Customize your streaming endpoint served from core tool in variable 'url' if different.
    url = 'http://localhost:7071/api/streaming_upload'
    file_path = r'<file path>'

    response = await stream_to_server(url, file_path)
    print(response)

if __name__ == "__main__":
    asyncio.run(main())

出力は、戻り値および出力パラメーターの両方で表現できます。 出力が 1 つのみの場合は、戻り値の使用をお勧めします。 複数の出力の場合は、出力パラメーターを使用する必要があります。

出力バインディングの値として関数の戻り値を使用するには、バインディングの name プロパティを function.json ファイル内の $return に設定する必要があります。

複数の出力を生成するには、azure.functions.Out インターフェイスによって提供される set() メソッドを使用して、バインディングに値を割り当てます。 たとえば、次の関数を使用すると、キューにメッセージをプッシュすることに加え、HTTP 応答を返すこともできます。

{
  "scriptFile": "__init__.py",
  "bindings": [
    {
      "name": "req",
      "direction": "in",
      "type": "httpTrigger",
      "authLevel": "anonymous"
    },
    {
      "name": "msg",
      "direction": "out",
      "type": "queue",
      "queueName": "outqueue",
      "connection": "STORAGE_CONNECTION_STRING"
    },
    {
      "name": "$return",
      "direction": "out",
      "type": "http"
    }
  ]
}

import azure.functions as func

def main(req: func.HttpRequest,
         msg: func.Out[func.QueueMessage]) -> str:

    message = req.params.get('body')
    msg.set(message)
    return message

出力は、戻り値および出力パラメーターの両方で表現できます。 出力が 1 つのみの場合は、戻り値の使用をお勧めします。 出力が複数の場合は、出力パラメーターを使用する必要があります。

複数の出力を生成するには、azure.functions.Out インターフェイスによって提供される set() メソッドを使用して、バインディングに値を割り当てます。 たとえば、次の関数を使用すると、キューにメッセージをプッシュすることに加え、HTTP 応答を返すこともできます。

# function_app.py
import azure.functions as func

app = func.FunctionApp()

@app.write_blob(arg_name="msg", path="output-container/{name}",
                connection="CONNECTION_STRING")
def test_function(req: func.HttpRequest,
                  msg: func.Out[str]) -> str:

    message = req.params.get('body')
    msg.set(message)
    return message

HTTP トリガーは function.json ファイルで定義されています。 バインディングの name は、関数の名前付きパラメーターと一致している必要があります。 前の例では、バインド名 req が使用されています。 このパラメーターは HttpRequest オブジェクトであり、HttpResponse オブジェクトが返されます。

HttpRequest オブジェクトからは、要求ヘッダー、クエリ パラメーター、ルート パラメーター、およびメッセージ本文を取得できます。

次の例は、Python 用の HTTP トリガー テンプレートからのものです。

def main(req: func.HttpRequest) -> func.HttpResponse:
    headers = {"my-http-header": "some-value"}

    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}!", headers=headers)
    else:
        return func.HttpResponse(
             "Please pass a name on the query string or in the request body",
             headers=headers, status_code=400
        )

この関数では、name クエリ パラメーターの値は HttpRequest オブジェクトの params パラメーターから取得されます。 JSON でエンコードされたメッセージ本文は get_json メソッドを使用して読み取ります。

同様に、返される status_code オブジェクトに応答メッセージの status_code および headers を設定できます。

HTTP トリガーは、HttpRequest オブジェクトである名前付きバインディング パラメーターを受け取り、HttpResponse オブジェクトを返すメソッドであると定義されます。 function_name デコレーターをメソッドに適用して関数名を定義しますが、HTTP エンドポイントは route デコレーターを適用することで設定します。

この例は、Python v2 プログラミング モデルの HTTP トリガー テンプレートのものであり、バインディング パラメーター名は req です。 これは、Azure Functions Core Tools または Visual Studio Code を使用して関数を作成するときに提供されるサンプル コードです。

@app.function_name(name="HttpTrigger1")
@app.route(route="hello")
def test_function(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
        )

HttpRequest オブジェクトからは、要求ヘッダー、クエリ パラメーター、ルート パラメーター、およびメッセージ本文を取得できます。 この関数では、name クエリ パラメーターの値は HttpRequest オブジェクトの params パラメーターから取得されます。 JSON でエンコードされたメッセージ本文は get_json メソッドを使用して読み取ります。

同様に、返される status_code オブジェクトに応答メッセージの status_code および headers を設定できます。

この例で名前を渡すには、関数の実行時に指定された URL を貼り付けてから、それに "?name={name}" を付加します。

Flask や FastAPI などの Web Server Gateway Interface (WSGI) 互換および Asynchronous Server Gateway Interface (ASGI) 互換フレームワークを、HTTP によってトリガーされる Python 関数と共に使用できます。 このセクションでは、これらのフレームワークをサポートするために関数を変更する方法を示します。

まず、次の例に示すように、function.json ファイルを更新して、HTTP トリガーに route を含める必要があります。

{
  "scriptFile": "__init__.py",
  "bindings": [
    {
       "authLevel": "anonymous",
       "type": "httpTrigger",
       "direction": "in",
       "name": "req",
       "methods": [
           "get",
           "post"
       ],
       "route": "{*route}"
    },
    {
       "type": "http",
       "direction": "out",
       "name": "$return"
    }
  ]
}

次の例に示すように、host.json ファイルを更新して、HTTP routePrefix を含める必要があります。

{
  "version": "2.0",
  "logging": 
  {
    "applicationInsights": 
    {
      "samplingSettings": 
      {
        "isEnabled": true,
        "excludedTypes": "Request"
      }
    }
  },
  "extensionBundle": 
  {
    "id": "Microsoft.Azure.Functions.ExtensionBundle",
    "version": "[3.*, 4.0.0)"
  },
  "extensions": 
  {
    "http": 
    {
        "routePrefix": ""
    }
  }
}

フレームワークで使用されるインターフェイスに応じて、Python コード ファイル init.py を更新します。 次の例では、ASGI ハンドラー アプローチまたは Flask の WSGI ラッパー アプローチのいずれかを示します。

app = fastapi.FastAPI()

@app.get("hello/{name}")
async def get_name(name: str):
  return {"name": name}

def main(req: func.HttpRequest, context: func.Context) -> func.HttpResponse:
    return func.AsgiMiddleware(app).handle(req, context)

app = Flask("Test")

@app.route("hello/<name>", methods=["GET"])
def hello(name: str):
    return f"hello {name}"

def main(req: func.HttpRequest, context) -> func.HttpResponse:
  logging.info("Python HTTP trigger function processed a request.")
  return func.WsgiMiddleware(app).handle(req, context)

Flask や FastAPI などの Asynchronous Server Gateway Interface (ASGI) 互換フレームワークおよび Web Server Gateway Interface (WSGI) 互換フレームワーク (Flask や FastAPI など) を、HTTP によってトリガーされる Python 関数と共に使用できます。 次の例に示すように、最初に host.json ファイルを更新して、HTTP routePrefix を含める必要があります。

{
  "version": "2.0",
  "logging": 
  {
    "applicationInsights": 
    {
      "samplingSettings": 
      {
        "isEnabled": true,
        "excludedTypes": "Request"
      }
    }
  },
  "extensionBundle": 
  {
    "id": "Microsoft.Azure.Functions.ExtensionBundle",
    "version": "[2.*, 3.0.0)"
  },
  "extensions": 
  {
    "http": 
    {
        "routePrefix": ""
    }
  }
}

フレームワーク コードは次の例のようになります。

# function_app.py

import azure.functions as func 
from fastapi import FastAPI, Request, Response 

fast_app = FastAPI() 

@fast_app.get("/return_http_no_body") 
async def return_http_no_body(): 
    return Response(content="", media_type="text/plain") 

app = func.AsgiFunctionApp(app=fast_app, 
                           http_auth_level=func.AuthLevel.ANONYMOUS) 

# function_app.py

import azure.functions as func 
from flask import Flask, Response 

flask_app = Flask(__name__) 

@flask_app.get("/return_http") 
def return_http(): 
    return Response("<h1>Hello World™</h1>", mimetype="text/html") 

app = func.WsgiFunctionApp(app=flask_app.wsgi_app, 
                           http_auth_level=func.AuthLevel.ANONYMOUS) 

Azure Functions では、サービス接続文字列などのアプリケーション設定は、実行時に環境変数として公開されます。 コードでこれらの設定にアクセスするには、主に 2 つの方法があります。

どちらの方法も、import os を宣言する必要があります。

次の設定では、os.environ["myAppSetting"] を使用して myAppSetting という名前のキーのos.environ["myAppSetting"]が取得されます。

import logging
import os

import azure.functions as func

def main(req: func.HttpRequest) -> func.HttpResponse:
  # Get the setting named 'myAppSetting'
  my_app_setting_value = os.environ["myAppSetting"]
  logging.info(f'My app setting value:{my_app_setting_value}')

ローカル開発の場合、アプリケーション設定は local.settings.json ファイルに保持されます。

Azure Functions では、サービス接続文字列などのアプリケーション設定は、実行時に環境変数として公開されます。 コードでこれらの設定にアクセスするには、主に 2 つの方法があります。

どちらの方法も、import os を宣言する必要があります。

次の設定では、os.environ["myAppSetting"] を使用して myAppSetting という名前のキーのos.environ["myAppSetting"]が取得されます。

import logging
import os

import azure.functions as func

app = func.FunctionApp()

@app.function_name(name="HttpTrigger1")
@app.route(route="req")
def main(req: func.HttpRequest) -> func.HttpResponse:
  # Get the setting named 'myAppSetting'
  my_app_setting_value = os.environ["myAppSetting"]
  logging.info(f'My app setting value:{my_app_setting_value}')

ローカル開発の場合、アプリケーション設定は local.settings.json ファイルに保持されます。

最初に、<project_root>/my_second_function/function.json ファイルを作成し、この関数を HTTP トリガーとして定義します。

{
  "scriptFile": "__init__.py",
  "entryPoint": "main",
  "bindings": [
    {
      "authLevel": "function",
      "type": "httpTrigger",
      "direction": "in",
      "name": "req",
      "methods": [
        "get",
        "post"
      ]
    },
    {
      "type": "http",
      "direction": "out",
      "name": "$return"
    }
  ]
}

次に、my_second_function と shared_code.my_second_helper_function を実装できます。

# <project_root>/my_second_function/__init__.py
import azure.functions as func
import logging

# Use absolute import to resolve shared_code modules
from shared_code import my_second_helper_function

# Define an HTTP trigger that accepts the ?value=<int> query parameter
# Double the value and return the result in HttpResponse
def main(req: func.HttpRequest) -> func.HttpResponse:
  logging.info('Executing my_second_function.')

  initial_value: int = int(req.params.get('value'))
  doubled_value: int = my_second_helper_function.double(initial_value)

  return func.HttpResponse(
    body=f"{initial_value} * 2 = {doubled_value}",
    status_code=200
    )

# <project_root>/shared_code/__init__.py
# Empty __init__.py file marks shared_code folder as a Python package

# <project_root>/shared_code/my_second_helper_function.py

def double(value: int) -> int:
  return value * 2

HTTP トリガーのテスト ケースの記述を開始できます。

# <project_root>/tests/test_my_second_function.py
import unittest

import azure.functions as func
from my_second_function import main

class TestFunction(unittest.TestCase):
  def test_my_second_function(self):
    # Construct a mock HTTP request.
    req = func.HttpRequest(method='GET',
                           body=None,
                           url='/api/my_second_function',
                           params={'value': '21'})
    # Call the function.
    resp = main(req)

    # Check the output.
    self.assertEqual(resp.get_body(), b'21 * 2 = 42',)

.venv の Python 仮想環境フォルダー内で、任意の Python テスト フレームワーク (pip install pytest など) をインストールします。 続いて pytest tests を実行して、テスト結果を確認します。

まず、<project_root>/function_app.py ファイルを作成し、my_second_function 関数 HTTP トリガーおよび shared_code.my_second_helper_function として実装します。

# <project_root>/function_app.py
import azure.functions as func
import logging

# Use absolute import to resolve shared_code modules
from shared_code import my_second_helper_function

app = func.FunctionApp()

# Define the HTTP trigger that accepts the ?value=<int> query parameter
# Double the value and return the result in HttpResponse
@app.function_name(name="my_second_function")
@app.route(route="hello")
def main(req: func.HttpRequest) -> func.HttpResponse:
    logging.info('Executing my_second_function.')

    initial_value: int = int(req.params.get('value'))
    doubled_value: int = my_second_helper_function.double(initial_value)

    return func.HttpResponse(
        body=f"{initial_value} * 2 = {doubled_value}",
        status_code=200
    )

# <project_root>/shared_code/__init__.py
# Empty __init__.py file marks shared_code folder as a Python package

# <project_root>/shared_code/my_second_helper_function.py

def double(value: int) -> int:
  return value * 2

HTTP トリガーのテスト ケースの記述を開始できます。

# <project_root>/tests/test_my_second_function.py
import unittest
import azure.functions as func

from function_app import main

class TestFunction(unittest.TestCase):
  def test_my_second_function(self):
    # Construct a mock HTTP request.
    req = func.HttpRequest(method='GET',
                           body=None,
                           url='/api/my_second_function',
                           params={'value': '21'})
    # Call the function.
    func_call = main.build().get_user_function()
    resp = func_call(req)
    # Check the output.
    self.assertEqual(
        resp.get_body(),
        b'21 * 2 = 42',
    )

.venv の Python 仮想環境フォルダー内で、任意の Python テスト フレームワーク (pip install pytest など) をインストールします。 続いて pytest tests を実行して、テスト結果を確認します。