LoginSignup
1
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

@FukuharaYohei(Fukuhara)inMicrosoft

Azure Durable FunctionsをPythonで実装

Last updated at Posted at 2025-07-13

Azure Durable Functionsを試した時のメモです。以下の記事のこの絵のパターンがやりたくて、試してみました。理由は、LLMを使う少し長めのStatusを持ったFunctionsで、自動でStatus管理してくれるのを見たかったからです。
image.png

主に以下の記事から試しました。

Step

0. 前提

「サポートされている言語」にはPython3.7以上として書いていないのですが、「クイックスタート」には「Python バージョン 3.7、3.8、3.9、または 3.10 がインストールされている。」と書かれているので、Python 3.10で試しました。3.11以降で動くかは確認していないです。

種類 Version 備考
OS Ubuntu22.04.5 LTS WSL2で動かしています
Python 3.10.6 少し古いです
Poetry 2.1.3 仮想環境の管理に使用

Python パッケージ

種類 Version 備考
azure-functions 1.23.0
azure-durable-functions 1.3.2

1. プログラム実装

1.1. 初期設定

コマンドパレットからプロジェクト作成
image.png

「クイックスタート」と同じ選択
image.png

1.2. メインプログラム

クイックスタートと以下の点を変更しています。

  • パラメータから関数名を取得しない(するとエラーが起きたので)
  • カスタムステータスを設定
  • HTTP RequestのBodyを読み込む(ただPrintするだけで不使用)
  • Activityをもう１つ試す
  • Activityにスリープ処理を追加(ステータス変化を追うため)

プログラム全体です。

function_app.py
import time
import azure.functions as func
import azure.durable_functions as df

myApp = df.DFApp(http_auth_level=func.AuthLevel.ANONYMOUS)

# An HTTP-triggered function with a Durable Functions client binding
@myApp.route(route="orchestrators/hello_orchestrator")
#@myApp.route(route="orchestrators/{functionName}")
@myApp.durable_client_input(client_name="client")
async def http_start(req: func.HttpRequest, client: df.DurableOrchestrationClient):
#    function_name = req.route_params.get('functionName')
#    instance_id = await client.start_new(function_name)
    payload = req.get_json()
    instance_id = await client.start_new("hello_orchestrator", client_input=payload)
    response = client.create_check_status_response(req, instance_id)
    return response

# Orchestrator
@myApp.orchestration_trigger(context_name="context")
def hello_orchestrator(context):
    payload = context.get_input()
    print(payload)
    context.set_custom_status("Started")  
    result1 = yield context.call_activity("hello", "Seattle")
    context.set_custom_status({"stage": 1, "progress": 33})
    result2 = yield context.call_activity("hello2", "Tokyo")
    context.set_custom_status({"stage": 2, "progress": 66})
    result3 = yield context.call_activity("hello", "London")
    context.set_custom_status("Completed")  

    return [result1, result2, result3]

# Activity
@myApp.activity_trigger(input_name="city")
def hello(city: str):
    time.sleep(10)
    return f"Hello {city}"

# Activity
@myApp.activity_trigger(input_name="city")
def hello2(city: str):
    time.sleep(10)
    return f"Good day {city}"

以下、個別の解説。

クライアント関数

処理のトリガーの受け取り口となる関数。クイックスタートの通りreq.route_params.get('functionName')を実装したらここでエラー。今回は関数固定なので動的に取得せずにclient.start_newに固定値hello_orchestratorを渡しています。

http_start
# An HTTP-triggered function with a Durable Functions client binding
@myApp.route(route="orchestrators/hello_orchestrator")
#@myApp.route(route="orchestrators/{functionName}")
@myApp.durable_client_input(client_name="client")
async def http_start(req: func.HttpRequest, client: df.DurableOrchestrationClient):
#    function_name = req.route_params.get('functionName')
#    instance_id = await client.start_new(function_name)
    payload = req.get_json()
    instance_id = await client.start_new("hello_orchestrator", client_input=payload)
    response = client.create_check_status_response(req, instance_id)
    return response

オーケストレーター関数

ここで実行する内容を書きます。set_custom_statusでカスタムステータスを設定。

hello_orchestrator
# Orchestrator
@myApp.orchestration_trigger(context_name="context")
def hello_orchestrator(context):
    payload = context.get_input()
    print(payload)
    context.set_custom_status("Started")  
    result1 = yield context.call_activity("hello", "Seattle")
    context.set_custom_status({"stage": 1, "progress": 33})
    result2 = yield context.call_activity("hello2", "Tokyo")
    context.set_custom_status({"stage": 2, "progress": 66})
    result3 = yield context.call_activity("hello", "London")
    context.set_custom_status("Completed")  

    return [result1, result2, result3]

1.3. ストレージ エミュレーター

ストレージ エミュレーター設定のため、local.settings.json を変更

local.settings.json
{
  "IsEncrypted": false,
  "Values": {
    "AzureWebJobsStorage": "UseDevelopmentStorage=true",
    "FUNCTIONS_WORKER_RUNTIME": "python"
  }
}

2. ローカルでのテスト

2.1. 起動

普通のFunctionと同様、コマンドパレットからAzurite を起動し、Functionsもfunc host start -wで起動

$ func host start -w
Found Python version 3.10.12 (python3).

Azure Functions Core Tools
Core Tools Version:       4.0.6821 Commit hash: N/A +c09a2033faa7ecf51b3773308283af0ca9a99f83 (64-bit)
Function Runtime Version: 4.1036.1.23224

[2025-07-13T06:32:14.029Z] Worker process started and initialized.

Functions:

        http_start:  http://localhost:7071/api/orchestrators/hello_orchestrator

        hello: activityTrigger

        hello2: activityTrigger

        hello_orchestrator: orchestrationTrigger

For detailed output, run func with --verbose flag.
[2025-07-13T06:32:18.975Z] Host lock lease acquired by instance ID '00000000000000000000000063AA5621'.

2.2. 実行

Curlでhttp://localhost:7071/api/orchestrators/hello_orchestratorにリクエスト投げると以下のResonseが返ります。これで実行完了です。JSONは見やすいように改行など整形しています(以下同じ)。

Response
$ curl -X POST   http://localhost:7071/api/orchestrators/hello_orchestrator   -H "Content-Type: application/json"   -d '{
    "city": "Tokyo",
    "count": 3
  }'
{
    "id": "ecd9dd0ace804c61b04bd94f7da9fdfc",
    "statusQueryGetUri": "http://localhost:7071/runtime/webhooks/durabletask/instances/ecd9dd0ace804c61b04bd94f7da9fdfc?taskHub=TestHubName&connection=Storage&code=BMPYKiPU07w6kBO6ZZL8RofyK2SHMhMa-XqZuv3KtKvIAzFuFhlOrQ==",
    "sendEventPostUri": "http://localhost:7071/runtime/webhooks/durabletask/instances/ecd9dd0ace804c61b04bd94f7da9fdfc/raiseEvent/{eventName}?taskHub=TestHubName&connection=Storage&code=BMPYKiPU07w6kBO6ZZL8RofyK2SHMhMa-XqZuv3KtKvIAzFuFhlOrQ==",
    "terminatePostUri": "http://localhost:7071/runtime/webhooks/durabletask/instances/ecd9dd0ace804c61b04bd94f7da9fdfc/terminate?reason={text}&taskHub=TestHubName&connection=Storage&code=BMPYKiPU07w6kBO6ZZL8RofyK2SHMhMa-XqZuv3KtKvIAzFuFhlOrQ==",
    "rewindPostUri": "http://localhost:7071/runtime/webhooks/durabletask/instances/ecd9dd0ace804c61b04bd94f7da9fdfc/rewind?reason={text}&taskHub=TestHubName&connection=Storage&code=BMPYKiPU07w6kBO6ZZL8RofyK2SHMhMa-XqZuv3KtKvIAzFuFhlOrQ==",
    "purgeHistoryDeleteUri": "http://localhost:7071/runtime/webhooks/durabletask/instances/ecd9dd0ace804c61b04bd94f7da9fdfc?taskHub=TestHubName&connection=Storage&code=BMPYKiPU07w6kBO6ZZL8RofyK2SHMhMa-XqZuv3KtKvIAzFuFhlOrQ==",
    "restartPostUri": "http://localhost:7071/runtime/webhooks/durabletask/instances/ecd9dd0ace804c61b04bd94f7da9fdfc/restart?taskHub=TestHubName&connection=Storage&code=BMPYKiPU07w6kBO6ZZL8RofyK2SHMhMa-XqZuv3KtKvIAzFuFhlOrQ==",
    "suspendPostUri": "http://localhost:7071/runtime/webhooks/durabletask/instances/ecd9dd0ace804c61b04bd94f7da9fdfc/suspend?reason={text}&taskHub=TestHubName&connection=Storage&code=BMPYKiPU07w6kBO6ZZL8RofyK2SHMhMa-XqZuv3KtKvIAzFuFhlOrQ==",
    "resumePostUri": "http://localhost:7071/runtime/webhooks/durabletask/instances/ecd9dd0ace804c61b04bd94f7da9fdfc/resume?reason={text}&taskHub=TestHubName&connection=Storage&co   後略"
}

statusQueryGetUriの値のURLにブラウザでアクセスします。何回かアクセスするとステータスが遷移していくのがわかります。

開始時
{
    "name": "hello_orchestrator",
    "instanceId": "ecd9dd0ace804c61b04bd94f7da9fdfc",
    "runtimeStatus": "Running",
    "input": "{\"city\": \"Tokyo\", \"count\": 3}",
    "customStatus": "Started",
    "output": null,
    "createdTime": "2025-07-13T06:35:23Z",
    "lastUpdatedTime": "2025-07-13T06:35:23Z"
}
途中(2つ目実行中)
{
    "name": "hello_orchestrator",
    "instanceId": "ecd9dd0ace804c61b04bd94f7da9fdfc",
    "runtimeStatus": "Running",
    "input": "{\"city\": \"Tokyo\", \"count\": 3}",
    "customStatus": {
        "stage": 1,
        "progress": 33
    },
    "output": null,
    "createdTime": "2025-07-13T06:44:19Z",
    "lastUpdatedTime": "2025-07-13T06:44:30Z"
}
完了時
{
    "name": "hello_orchestrator",
    "instanceId": "ecd9dd0ace804c61b04bd94f7da9fdfc",
    "runtimeStatus": "Completed",
    "input": "{\"city\": \"Tokyo\", \"count\": 3}",
    "customStatus": "Completed",
    "output": [
        "Hello Seattle",
        "Good day Tokyo",
        "Hello London"
    ],
    "createdTime": "2025-07-13T06:44:19Z",
    "lastUpdatedTime": "2025-07-13T06:44:50Z"
}

多分、動き同じなので、今回はデプロイまでは検証しないです。

1
0
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
1
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?