Rychlý start: Vytvoření aplikace Python Durable Functions

Pomocí Durable Functions, funkce Azure Functions, můžete psát stavové bezserverové pracovní postupy v Python. V tomto rychlém startu naklonujete a spustíte ukázkovou aplikaci, která ukazuje dva běžné vzory orchestrace:

  • Řetězení funkcí: Volá aktivity postupně (Tokio → Seattle → Londýn).
  • Fan-out/fan-in: Volá aktivity paralelně napříč pěti městy a pak agreguje výsledky.

Na konci budete mít obě orchestrace spuštěné lokálně pomocí emulátoru Durable Task Scheduler a budete moct zobrazit jejich stav na dashboardu.

  • Naklonujte a připravte ukázkový projekt Hello Cities.
  • Nastavte emulátor plánovače úloh Durable a Azurite pro místní vývoj.
  • Spusťte aplikaci funkcí a aktivujte obě orchestrace.
  • Zkontrolujte stav orchestrace a výstup na řídicím panelu Plánovače úloh Durable.

Předpoklady

Nastavení emulátoru Plánovače úloh Durable

Emulátor plánovače úloh Durable Task Scheduler poskytuje místní vývojové prostředí, abyste mohli otestovat orchestrace bez předplatného Azure. Hostitel Functions také vyžaduje Azurite pro místní úložiště.

Spusťte oba kontejnery:

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

Tip

Jakmile je emulátor spuštěný, můžete přistupovat k řídicímu panelu Durable Task Scheduler na adrese http://localhost:8082, kde můžete sledovat orchestrace.

Spusťte ukázku pro rychlý start

  1. Přejděte do ukázkového adresáře Hello Cities:

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

  2. Vytvořte virtuální prostředí a nainstalujte závislosti:

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

  3. Ověřte, že local.settings.json soubor obsahuje následující konfiguraci:

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

  4. Spusťte aplikaci funkcí:

    func start

  5. V samostatném terminálu aktivujte orchestraci řetězení funkcí :

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

    Odpověď obsahuje stavové adresy URL pro instanci orchestrátoru. statusQueryGetUri Zkopírujte hodnotu a spusťte ji a zkontrolujte výsledek:

    Invoke-RestMethod -Uri $response.statusQueryGetUri

  6. Spusťte orchestraci fan-out/fan-in:

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

Očekávaný výstup

Požadavek POST vrátí odpověď JSON s adresami URL stavu. Příklad:

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

Když zadáte dotaz na statusQueryGetUri a hodnota runtimeStatus orchestrace je Completed, výsledky pozdravu najdete v poli output. Orchestrace zřetězení vrátí:

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

Orchestrace fan-out/fan-in vrací:

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

Tip

Pokud se na runtimeStatus zobrazí Running nebo Pending, chvíli počkejte a znovu odešlete dotaz na statusQueryGetUri.

Otevřete řídicí panel Plánovače úloh Durable na adrese http://localhost:8082, abyste zobrazili stav orchestrace a historii spuštění.

Vysvětlení kódu

Ukázka používá programovací model Python v2 s dekorátory, kde jsou všechny funkce definovány v jednom souboru (function_app.py).

Funkce aktivity

Aktivita say_hello přebírá název města a vrací pozdrav:

@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}!"

Funkce orchestratoru

Orchestrátor řetězení postupně volá say_hello pro tři města:

@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]

Orchestrátor fan-out/fan-in plánuje aktivity paralelně:

@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

Klientské funkce

Funkce klienta aktivované protokolem HTTP spouští každou orchestraci. Například začátek řetězení:

@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)

Konfigurace

Ukázka používá emulátor Plánovače úloh Durable jako back-end úložiště. Toto nastavení je nakonfigurováno v 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)"
  }
}

Vyčistěte zdroje

Po dokončení zastavte kontejnery emulátoru:

docker stop dtsemulator azurite && docker rm dtsemulator azurite

Deaktivace virtuálního prostředí Python:

deactivate

Další kroky

  • Last updated on