Démarrage rapide : Créer une application Python Durable Functions

Utilisez Durable Functions, une fonctionnalité de Azure Functions, pour écrire des flux de travail serverless avec état dans Python. Dans ce guide de démarrage rapide, vous clonez et exécutez un exemple d’application qui illustre deux modèles d’orchestration courants :

Chaînage de fonctions : appelle les activités de manière séquentielle (Tokyo → Seattle → Londres).
Fan-out/fan-in : appelle des activités en parallèle sur cinq villes, puis agrège les résultats.

À la fin, vous aurez exécuté localement les deux orchestrations avec l’émulateur Durable Task Scheduler et pourrez voir leur état dans le tableau de bord.

Clonez et préparez l’exemple de projet Hello Cities.
Configurez l’émulateur Durable Task Scheduler et Azurite pour le développement local.
Exécutez l’application de fonctions et déclenchez les deux orchestrations.
Consultez l’état et les résultats de l’orchestration dans le tableau de bord de Durable Task Scheduler.

Prerequisites

Python 3.9+ installé.
Azure Functions Core Tools v4 ou version ultérieure.
Docker pour exécuter l’émulateur et Azurite.
Clonez le référentiel GitHub Durable Task Scheduler pour utiliser l’exemple de démarrage rapide.

Configurer l’émulateur Durable Task Scheduler

L’émulateur Durable Task Scheduler fournit un environnement de développement local afin de pouvoir tester des orchestrations sans abonnement Azure. L’hôte Functions nécessite également Azurite pour le stockage local.

Démarrez les deux conteneurs :

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

Une fois l’émulateur en cours d’exécution, vous pouvez accéder au tableau de bord Durable Task Scheduler à l’adresse http://localhost:8082 pour surveiller les orchestrations.

Exécuter l’exemple de démarrage rapide

Accédez à l’exemple de répertoire Hello Cities :
```
cd samples/durable-functions/python/hello-cities
```

Créez un environnement virtuel et installez les dépendances :

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

Vérifiez que le local.settings.json fichier contient la configuration suivante :

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

Démarrez l’application de fonctions :
```
func start
```
Dans un terminal distinct, déclenchez l’orchestration de chaînage de fonctions :
```
$response = Invoke-RestMethod -Method POST -Uri http://localhost:7071/api/StartChaining
$response
```
La réponse contient des URL d’état pour l’instance d’orchestration. Copiez la valeur et exécutez-la statusQueryGetUri pour vérifier le résultat :
```
Invoke-RestMethod -Uri $response.statusQueryGetUri
```

Déclenchez l’orchestration fan-out/fan-in :

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

Sortie attendue

La requête POST retourne une réponse JSON avec des URL d’état. Par exemple:

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

Lorsque vous interrogez statusQueryGetUri et que le runtimeStatus de l’orchestration est Completed, vous trouverez les résultats du message d’accueil dans le champ output. L’orchestration en chaîne renvoie :

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

L’orchestration fan-out/fan-in renvoie :

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

Tip

Si runtimeStatus affiche Running ou Pending, attendez un instant et interrogez de nouveau statusQueryGetUri.

Ouvrez le tableau de bord Durable Task Scheduler à l’adresse http://localhost:8082 pour afficher l’état de l’orchestration et l’historique d’exécution.

Comprendre le code

L’exemple utilise le modèle de programmation Python v2 avec des décorateurs, où toutes les fonctions sont définies dans un fichier unique (function_app.py).

Fonction d’activité

L’activité say_hello prend un nom de ville et retourne un message d’accueil :

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

Fonctions d’orchestrateur

L’orchestrateur de chaînage appelle say_hello de manière séquentielle pour trois villes :

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

L’orchestrateur fan-out/fan-in planifie les activités en parallèle :

@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

Fonctions du client

Les fonctions clientes déclenchées par HTTP démarrent chaque orchestration. Par exemple, le module de démarrage de l’enchaînement :

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

Configuration

L’exemple utilise l’émulateur Durable Task Scheduler comme back-end de stockage. Ceci est configuré dans 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)"
  }
}

Nettoyer les ressources

Arrêtez les conteneurs de l’émulateur lorsque vous avez terminé :

docker stop dtsemulator azurite && docker rm dtsemulator azurite

Pour désactiver l’environnement virtuel Python :

deactivate

Étapes suivantes

Découvrez les modèles d’application common Durable Functions.
Découvrez les fournisseurs de stockage Durable Functions.

Commentaires

Est-ce que cette page vous a été utile?

Last updated on 2026-06-05