はじめに
前回の記事からの続きです。Azure FunctionsとCosmos DBを組み合わせてイベント駆動型の小規模なアプリケーションを作ってみましょう。
HTTPトリガー、タイマートリガーを紹介し、最後にCosmosDBのレコードが更新されたことをトリガーとする変更フィードという仕組みを使ってアプリケーションを作ってみます。
VSCodeの拡張機能インストール
azure databases 拡張機能のインストール
Azure Cosmos DBアカウントの作成
「Azure Databases: Create Server…」を選択する。
以下の通り作成する。
作成完了。
Cosmos DBを作成する
「azure」タブから先ほど作成したcosmos DBを選択し「create database」を選択する。
下記の情報を入力する。
Cosmos DBアカウントを選択し「Copy Connection String」で接続情報をコピーする。
以降の細かい手順は以下に沿って実施する。
流れとしては、CosmosDBの接続情報をローカルにダウンロードしてきて、ローカル⇔CosmosDBも接続しつつ、デプロイ環境⇔CosmosDBもつなぐ。
コード
import azure.functions as func
import logging
app = func.FunctionApp()
@app.function_name(name="HttpTrigger1")
@app.route(route="hello", auth_level=func.AuthLevel.ANONYMOUS)
@app.queue_output(arg_name="msg", queue_name="outqueue", connection="AzureWebJobsStorage")
@app.cosmos_db_output(arg_name="outputDocument", database_name="my-database", container_name="my-container", connection="CosmosDbConnectionSetting")
def test_function(req: func.HttpRequest, msg: func.Out[func.QueueMessage],
outputDocument: func.Out[func.Document]) -> func.HttpResponse:
logging.info('Python HTTP trigger function processed a request.')
logging.info('Python Cosmos DB 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:
outputDocument.set(func.Document.from_dict({"id": name}))
msg.set(name)
return func.HttpResponse(f"Hello {name}!")
else:
return func.HttpResponse(
"Please pass a name on the query string or in the request body",
status_code=400
)
動作確認
HTTPリクエストにnameパラメータが入っていると、そのパラメータに入っている値をCosmos DBに登録することができるようになる。
タイマートリガーにしてみる
HTTPトリガーではなく、定期的に自動実行してCosmosDBに登録してくれるタイマートリガーにしてみる。
コード
import logging
import azure.functions as func
import datetime
app = func.FunctionApp()
@app.schedule(schedule="0 */1 * * * *", arg_name="myTimer", run_on_startup=True,
use_monitor=False)
@app.cosmos_db_output(arg_name="outputDocument", database_name="my-database", container_name="my-container", connection="CosmosDbConnectionSetting")
def timer_trigger01(myTimer: func.TimerRequest, outputDocument: func.Out[func.Document]) -> None:
if myTimer.past_due:
logging.info('The timer is past due!')
current_time = datetime.datetime.utcnow().isoformat()
logging.info('Python timer trigger function executed.')
outputDocument.set(func.Document.from_dict({"id": current_time}))
VSCodeでAzure Functions作成時に、既定のタイマートリガー関数を作成してから以下の変更を加えた。
- 関数の引数に「OutputDocument」を追加。なお「→ None」の部分は返り値の型なので、タイマートリガーの場合はNoneでよい
- outputDocument.set()関数を追加
カンタン！
動作確認
ちゃんとCosmosDBに値が入っている。
Cosmos DB triggerにしてみる
Azure FunctionsのCosmos DB Triggerを使えば、指定したCosmos DB内の特定のテーブルのレコードが変更された場合に関数を実行することができるようになります（変更フィード）。
参考
コード
import azure.functions as func
import logging
app = func.FunctionApp()
@app.cosmos_db_trigger(arg_name="azcosmosdb", container_name="my-container",
database_name="my-database", connection="azfoocosmosdb02_DOCUMENTDB",
LeaseContainerName = "leases", CreateLeaseContainerIfNotExists = "true")
def cosmosdb_trigger03(azcosmosdb: func.DocumentList):
logging.info('Python CosmosDB triggered.')
VSCodeでAzure Functions作成時に、既定のCosmos Trigger関数を作成してから以下の変更を加えた。
- 「LeaseContainerName = "leases", CreateLeaseContainerIfNotExists = "true"」をデフォルトのコードに追加（しないと以下のエラーになる）
エラー
This builder instance has already been used to build a processor. Create a new instance to
build another.
また、local.setteing.jsonの内容はあくまでもローカルでの実行時の設定値であるため、デプロイしたときは「azfoocosmosdb02_DOCUMENTDB」とその値をF1キー→「Azure functions : add new settings」から追加する必要はあることに注意（ポータルからも追加可能）。
おわりに
いかがでしたでしょうか。Azure FunctionsとCosmos DBの様々な機能を使えば、作りたい基盤を簡単に構築することができます。
面倒な部分はAzureに任せて、アプリケーションの製造に集中しましょう！