Cosmos DBトリガーを試す

azure/python-function-example/compose.yaml

@@ -0,0 +1,6 @@
+services:
+  cosmos:
+    image: mcr.microsoft.com/cosmosdb/linux/azure-cosmos-emulator
+    ports:
+      - "8081:8081"
+      - "10250-10255:10250-10255"

azure/python-function-example/function_app.py

@@ -1,5 +1,3 @@
-import datetime
-import json
 import logging
 
 import azure.functions as func
@@ -29,3 +27,19 @@ def HttpExample(req: func.HttpRequest) -> 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,
         )
+
+
+@app.function_name(name="CosmosDBTrigger")
+@app.cosmos_db_trigger(
+    arg_name="documents",
+    connection="CosmosDBConnectionString",
+    database_name="example_db",
+    container_name="example_container",
+    create_lease_container_if_not_exists=True,
+)
+def test_function(documents: func.DocumentList) -> None:
+    if documents:
+        for index, doc in enumerate(documents):
+            logging.info(f"{index} Document: {dict(doc)}")
+    else:
+        logging.info("No documents found")