Schnellstart: Erstellen einer Python Durable Functions-App

Verwenden Sie Durable Functions, ein Feature von Azure Functions, um zustandslose serverlose Workflows in Python zu schreiben. In dieser Schnellstartanleitung klonen und führen Sie eine Beispiel-App aus, die zwei gängige Orchestrierungsmuster veranschaulicht:

• Funktionskette: Ruft Aktivitäten sequenziell auf (Tokio → Seattle → London).
• Fan-out/Fan-In: Ruft Aktivitäten parallel in fünf Städten auf und aggregiert dann die Ergebnisse.

Am Ende haben Sie beide Orchestrierungen lokal mit dem Emulator " Durable Task Scheduler " ausgeführt und können ihren Status im Dashboard anzeigen.

Voraussetzungen

• Python 3,9+ installiert.
• Azure Functions Core Tools v4 oder höher.
• Docker für die Ausführung des Emulators und der Azurite.
• Klonen Sie den Durable Task Scheduler GitHub Repository, um das Schnellstartbeispiel zu verwenden.

Einrichten des Emulators "Durable Task Scheduler"

Der Emulator Durable Task Scheduler stellt eine lokale Entwicklungsumgebung bereit, sodass Sie Orchestrierungen ohne Azure Abonnement testen können. Der Functions-Host erfordert auch Azurite für den lokalen Speicher.

Starten Sie beide Container:

docker run -d --name dtsemulator -p 8080:8080 -p 8082:8082 \
  mcr.microsoft.com/dts/dts-emulator:latest

docker run -d --name azurite -p 10000:10000 -p 10001:10001 -p 10002:10002 \
  mcr.microsoft.com/azure-storage/azurite

Ausführen des Schnellstartbeispiels

1. Navigieren Sie zum Beispielverzeichnis "Hello Cities":

   cd samples/durable-functions/python/hello-cities
   

2. Erstellen Sie eine virtuelle Umgebung, und installieren Sie Abhängigkeiten:

   python -m venv .venv
   .venv\Scripts\activate
   pip install -r requirements.txt
   

3. Stellen Sie sicher, dass die local.settings.json Datei die folgende Konfiguration enthält:

   {
     "IsEncrypted": false,
     "Values": {
       "AzureWebJobsStorage": "UseDevelopmentStorage=true",
       "FUNCTIONS_WORKER_RUNTIME": "python",
       "DURABLE_TASK_SCHEDULER_CONNECTION_STRING": "Endpoint=http://localhost:8080;TaskHub=default;Authentication=None"
     }
   }
   

4. Starten Sie die Funktions-App:

   func start
   

5. Starten Sie in einem separaten Terminal die Orchestrierung der Verkettung von Funktionen:

   $response = Invoke-RestMethod -Method POST -Uri http://localhost:7071/api/StartChaining
   $response
   

   Die Antwort enthält Status-URLs für die Orchestrierungsinstanz. Kopieren Sie den statusQueryGetUri Wert, und führen Sie ihn aus, um das Ergebnis zu überprüfen:

   Invoke-RestMethod -Uri $response.statusQueryGetUri
   

6. Auslösen der Fanout-/Fan-In-Orchestrierung :

   $response = Invoke-RestMethod -Method POST -Uri http://localhost:7071/api/StartFanOutFanIn
   Invoke-RestMethod -Uri $response.statusQueryGetUri
   

Erwartete Ausgabe

Die POST-Anforderung gibt eine JSON-Antwort mit Status-URLs zurück. Zum Beispiel:

{
  "id": "<instanceId>",
  "statusQueryGetUri": "http://localhost:7071/runtime/webhooks/durabletask/instances/<instanceId>?code=...",
  "sendEventPostUri": "...",
  "terminatePostUri": "...",
  "purgeHistoryDeleteUri": "..."
}

Wenn Sie statusQueryGetUri abfragen und der runtimeStatus der Orchestrierung Completed ist, finden Sie die Begrüßungsergebnisse im Feld output. Die Verkettungs-Orchestrierung gibt Folgendes zurück:

{
  "name": "chaining_orchestration",
  "runtimeStatus": "Completed",
  "output": ["Hello Tokyo!", "Hello Seattle!", "Hello London!"]
}

Die Fan-out/Fan-in-Orchestrierung liefert Folgendes zurück:

{
  "name": "fan_out_fan_in_orchestration",
  "runtimeStatus": "Completed",
  "output": ["Hello Tokyo!", "Hello Seattle!", "Hello London!", "Hello Paris!", "Hello Berlin!"]
}

Öffnen Sie das Durable Task Scheduler-Dashboard unter http://localhost:8082, um den Orchestrierungsstatus und den Ausführungsverlauf einzusehen.

Grundlegendes zum Code

Im Beispiel wird das Python v2-Programmiermodell mit Dekoratoren verwendet, wobei alle Funktionen in einer einzelnen Datei definiert sind (function_app.py).

Aktivitätsfunktion

Die say_hello Aktivität nimmt einen Stadtnamen und gibt eine Begrüßung zurück:

@app.activity_trigger(input_name="city")
def say_hello(city: str) -> str:
    """Activity function that returns a greeting for a city."""
    logging.info(f"Saying hello to {city}.")
    return f"Hello {city}!"

Orchestratorfunktionen

Der Verkettungs-Orchestrator ruft sequenziell für drei Städte auf say_hello :

@app.orchestration_trigger(context_name="context")
def chaining_orchestration(context: df.DurableOrchestrationContext):
    """Function chaining orchestration: calls activities sequentially."""
    result1 = yield context.call_activity("say_hello", "Tokyo")
    result2 = yield context.call_activity("say_hello", "Seattle")
    result3 = yield context.call_activity("say_hello", "London")
    return [result1, result2, result3]

Der Fan-out/Fan-in-Orchestrator plant Aktivitäten parallel:

@app.orchestration_trigger(context_name="context")
def fan_out_fan_in_orchestration(context: df.DurableOrchestrationContext):
    """Fan-out/Fan-in orchestration: calls activities in parallel."""
    cities = ["Tokyo", "Seattle", "London", "Paris", "Berlin"]

    # Fan-out: schedule all activities in parallel
    parallel_tasks = []
    for city in cities:
        task = context.call_activity("say_hello", city)
        parallel_tasks.append(task)

    # Fan-in: wait for all to complete
    results = yield context.task_all(parallel_tasks)
    return results

Clientfunktionen

Durch HTTP ausgelöste Clientfunktionen starten jede Orchestrierung. Beispiel: Der Verkettungsstarter:

@app.route(route="StartChaining", methods=["POST"])
@app.durable_client_input(client_name="client")
async def start_chaining(req: func.HttpRequest, client) -> func.HttpResponse:
    """HTTP trigger to start the function chaining orchestration."""
    instance_id = await client.start_new("chaining_orchestration")
    logging.info(f"Started chaining orchestration with ID = '{instance_id}'.")
    return client.create_check_status_response(req, instance_id)

Konfiguration

Im Beispiel wird der Emulator "Durable Task Scheduler" als Speicher-Back-End verwendet. Dies ist konfiguriert in host.json:

{
  "version": "2.0",
  "logging": {
    "logLevel": {
      "DurableTask.Core": "Warning"
    }
  },
  "extensions": {
    "durableTask": {
      "hubName": "default",
      "storageProvider": {
        "type": "azureManaged",
        "connectionStringName": "DURABLE_TASK_SCHEDULER_CONNECTION_STRING"
      }
    }
  },
  "extensionBundle": {
    "id": "Microsoft.Azure.Functions.ExtensionBundle",
    "version": "[4.*, 5.0.0)"
  }
}

Bereinigen von Ressourcen

Beenden Sie die Emulatorcontainer, wenn Sie fertig sind:

docker stop dtsemulator azurite && docker rm dtsemulator azurite

So deaktivieren Sie die Python virtuelle Umgebung:

deactivate

Nächste Schritte

• Erfahren Sie mehr über common Durable Functions App-Muster.
• Erfahren Sie mehr über Durable Functions Speicheranbieter.