Chaînage de fonctions

• Suivez l’article sur le démarrage rapide
• Clone ou téléchargez le projet d’exemples à partir de GitHub

• Suivez l’article sur le démarrage rapide
• Clone ou téléchargez le projet d’exemples à partir de GitHub

• Suivez l’article sur le démarrage rapide
• Clone ou téléchargez le projet d’exemples à partir de GitHub

• Suivez l’article sur le démarrage rapide

[FunctionName("E1_HelloSequence")]
public static async Task<List<string>> Run(
    [OrchestrationTrigger] IDurableOrchestrationContext context)
{
    var outputs = new List<string>();

    outputs.Add(await context.CallActivityAsync<string>("E1_SayHello", "Tokyo"));
    outputs.Add(await context.CallActivityAsync<string>("E1_SayHello", "Seattle"));
    outputs.Add(await context.CallActivityAsync<string>("E1_SayHello_DirectInput", "London"));

    // returns ["Hello Tokyo!", "Hello Seattle!", "Hello London!"]
    return outputs;
}

Toutes les fonctions d’orchestration C# doivent avoir un paramètre de type DurableOrchestrationContext, qui existe dans l’assembly Microsoft.Azure.WebJobs.Extensions.DurableTask. Cet objet de contexte vous permet d’appeler d’autres fonctions d’activité et de passer les paramètres d’entrée à l’aide de sa méthode CallActivityAsync.

Le code appelle trois fois E1_SayHello en séquence avec des valeurs de paramètre différentes. La valeur renvoyée de chaque appel est ajoutée à la liste outputs, qui est retournée à la fin de la fonction.

function.json

Si vous utilisez Visual Studio Code ou le portail Azure pour le développement, voici le contenu du fichier function.json pour la fonction orchestrator. La plupart des fichiers function.json d’orchestrateur ressemblent presque exactement à cela.

{
  "scriptFile": "__init__.py",
  "bindings": [
    {
      "name": "context",
      "type": "orchestrationTrigger",
      "direction": "in"
    }
  ]
}

Le point essentiel est le type de liaison orchestrationTrigger. Toutes les fonctions d’orchestrateur doivent utiliser ce type de déclencheur.

__init__.py

Voici la fonction d’orchestrateur :

import azure.functions as func
import azure.durable_functions as df


def orchestrator_function(context: df.DurableOrchestrationContext):
    result1 = yield context.call_activity('E1_SayHello', "Tokyo")
    result2 = yield context.call_activity('E1_SayHello', "Seattle")
    result3 = yield context.call_activity('E1_SayHello', "London")
    return [result1, result2, result3]

main = df.Orchestrator.create(orchestrator_function)

Toutes les fonctions d’orchestration Python doivent inclure le package durable-functions. Il s'agit d'une bibliothèque qui vous permet d'écrire des Durable Functions dans Python. Deux différences clés entre une fonction d’orchestrateur et d’autres fonctions Python :

1. La fonction d’orchestrateur est une fonction de générateur.
2. Le fichier inscrit la fonction d’orchestrateur en indiquant main = df.Orchestrator.create(<orchestrator function name>) à la fin du fichier. Cela permet de la distinguer des autres fonctions d’assistance déclarées dans le fichier.

Cet objet context vous permet d’appeler d’autres fonctions d’activité et de transmettre les paramètres d’entrée à l’aide de sa méthode call_activity. Le code appelle E1_SayHello trois fois à la suite avec différentes valeurs de paramètre, en se servant de yield pour indiquer que l’exécution devrait attendre que les appels de fonction d'activité asynchrone soient complétés. La valeur renvoyée de chaque appel est restituée à la fin de la fonction.

L’exemple PowerShell n’est pas encore disponible.

Java exemple n'est pas encore disponible.

Ce code montre un orchestrateur qui appelle trois activités dans la séquence et transmet chaque sortie à l’activité suivante :

using Microsoft.DurableTask;

[DurableTask]
public class GreetingOrchestration : TaskOrchestrator<string, string>
{
    public override async Task<string> RunAsync(TaskOrchestrationContext context, string name)
    {
        // Step 1: Say hello to the person
        string greeting = await context.CallActivityAsync<string>(nameof(SayHelloActivity), name);

        // Step 2: Process the greeting
        string processedGreeting = await context.CallActivityAsync<string>(nameof(ProcessGreetingActivity), greeting);

        // Step 3: Finalize the response
        string finalResponse = await context.CallActivityAsync<string>(nameof(FinalizeResponseActivity), processedGreeting);

        return finalResponse;
    }
}

Tous les orchestrateurs .NET héritent de TaskOrchestrator<TInput, TOutput>. Avec le TaskOrchestrationContext, vous pouvez appeler des activités à l’aide de CallActivityAsync. Le code appelle trois activités dans la séquence, où chaque activité reçoit la sortie de l’activité précédente.

Le code suivant montre un orchestrateur qui appelle trois activités dans la séquence :

import {
  OrchestrationContext,
  TOrchestrator,
} from "@microsoft/durabletask-js";

const greetingOrchestrator: TOrchestrator = async function* (
  ctx: OrchestrationContext,
  name: string
): any {
  // Step 1: Say hello to the person
  const greeting: string = yield ctx.callActivity(sayHello, name);

  // Step 2: Process the greeting
  const processedGreeting: string = yield ctx.callActivity(processGreeting, greeting);

  // Step 3: Finalize the response
  const finalResponse: string = yield ctx.callActivity(finalizeResponse, processedGreeting);

  return finalResponse;
};

Tous les orchestrateurs JavaScript sont des fonctions de générateur (async function*) qui utilisent yield pour appeler des activités. La méthode callActivity du contexte d’orchestration planifie l’exécution de l’activité. Le code appelle trois activités dans l'ordre, et passe la sortie de chaque activité à la suivante.

Le code suivant montre un orchestrateur qui appelle trois activités dans la séquence :

from durabletask import task

# Orchestrator function
def function_chaining_orchestrator(ctx: task.OrchestrationContext, name: str) -> str:
    """Orchestrator that demonstrates function chaining pattern."""

    # Call first activity
    greeting = yield ctx.call_activity(say_hello, input=name)

    # Call second activity with the result from first activity
    processed_greeting = yield ctx.call_activity(process_greeting, input=greeting)

    # Call third activity with the result from second activity
    final_response = yield ctx.call_activity(finalize_response, input=processed_greeting)

    return final_response

Tous les orchestrateurs Python sont des fonctions de générateur qui utilisent yield pour appeler des activités. La méthode call_activity du contexte d’orchestration planifie l’exécution des activités. Le code appelle trois activités dans l'ordre, et passe la sortie de chaque activité à la suivante.

Cet exemple s’affiche pour .NET, JavaScript, Java et Python.

Le code suivant montre un orchestrateur qui appelle trois activités dans la séquence :

import com.microsoft.durabletask.*;

DurableTaskGrpcWorker worker = DurableTaskSchedulerWorkerExtensions.createWorkerBuilder(connectionString)
    .addOrchestration(new TaskOrchestrationFactory() {
        @Override
        public String getName() { return "ActivityChaining"; }

        @Override
        public TaskOrchestration create() {
            return ctx -> {
                String input = ctx.getInput(String.class);

                // Call activities in sequence, passing output from one to the next
                String x = ctx.callActivity("Reverse", input, String.class).await();
                String y = ctx.callActivity("Capitalize", x, String.class).await();
                String z = ctx.callActivity("ReplaceWhitespace", y, String.class).await();

                ctx.complete(z);
            };
        }
    })
    .build();

Dans Java, les orchestrateurs sont définis à l’aide de TaskOrchestrationFactory. La méthode du contexte planifie l’exécution de callActivity l’activité et await() attend le résultat. Le code appelle trois activités dans l'ordre, et passe la sortie de chaque activité à la suivante.

[FunctionName("E1_SayHello")]
public static string SayHello([ActivityTrigger] IDurableActivityContext context)
{
    string name = context.GetInput<string>();
    return $"Hello {name}!";
}

Les activités utilisent l’attribut ActivityTrigger. Utiliser IDurableActivityContext pour les actions d’activité, telles que la lecture d’entrée avec GetInput<T>.

E1_SayHello met en forme une chaîne de salutation.

Au lieu de lier à IDurableActivityContext, liez directement au type passé dans la fonction d'activité. Par exemple :

[FunctionName("E1_SayHello_DirectInput")]
public static string SayHelloDirectInput([ActivityTrigger] string name)
{
    return $"Hello {name}!";
}

E1_SayHello/function.json

Le fichier function.json pour la fonction de l’activité E1_SayHello est similaire à celle de E1_HelloSequence sauf qu’elle utilise un type de liaison activityTrigger à la place d’un type de liaison orchestrationTrigger.

{
  "scriptFile": "__init__.py",
  "bindings": [
    {
      "name": "name",
      "type": "activityTrigger",
      "direction": "in"
    }
  ]
}

L’implémentation de E1_SayHello est une opération de mise en forme de chaîne relativement simple.

E1_SayHello/__init__.py

def main(name: str) -> str:
    return f"Hello {name}!"

Contrairement à la fonction d’orchestrateur, une fonction d’activité ne nécessite aucune configuration particulière. L’entrée qui lui est transmise par la fonction d’orchestrateur est directement accessible en tant que paramètre de la fonction.

Exemple PowerShell bientôt disponible.

Java échantillon sera bientôt disponible.

Les activités du SDK Durable Task héritent de TaskActivity<TInput, TOutput>:

using Microsoft.DurableTask;
using Microsoft.Extensions.Logging;

[DurableTask]
public class SayHelloActivity : TaskActivity<string, string>
{
    private readonly ILogger<SayHelloActivity> _logger;

    public SayHelloActivity(ILogger<SayHelloActivity> logger)
    {
        _logger = logger;
    }

    public override Task<string> RunAsync(TaskActivityContext context, string name)
    {
        _logger.LogInformation("Activity SayHello called with name: {Name}", name);
        return Task.FromResult($"Hello {name}!");
    }
}

[DurableTask]
public class ProcessGreetingActivity : TaskActivity<string, string>
{
    public override Task<string> RunAsync(TaskActivityContext context, string greeting)
    {
        return Task.FromResult($"{greeting} How are you today?");
    }
}

[DurableTask]
public class FinalizeResponseActivity : TaskActivity<string, string>
{
    public override Task<string> RunAsync(TaskActivityContext context, string response)
    {
        return Task.FromResult($"{response} I hope you're doing well!");
    }
}

Utilisez l’injection de dépendances pour obtenir des services comme ILogger. Ajoutez l’attribut [DurableTask] pour inscrire l’activité auprès du worker.

Les activités du Kit de développement logiciel (SDK) Durable Task sont des fonctions simples :

import { ActivityContext } from "@microsoft/durabletask-js";

const sayHello = async (_ctx: ActivityContext, name: string): Promise<string> => {
  return `Hello ${name}!`;
};

const processGreeting = async (_ctx: ActivityContext, greeting: string): Promise<string> => {
  return `${greeting} How are you today?`;
};

const finalizeResponse = async (_ctx: ActivityContext, response: string): Promise<string> => {
  return `${response} I hope you're doing well!`;
};

Contrairement aux orchestrateurs, les activités peuvent effectuer des opérations d’E/S telles que les appels HTTP, les requêtes de base de données et l’accès aux fichiers. L’entrée est transmise directement en tant que paramètre.

Les activités du Kit de développement logiciel (SDK) Durable Task sont des fonctions simples :

from durabletask import task

def say_hello(ctx: task.ActivityContext, name: str) -> str:
    """First activity that greets the user."""
    return f"Hello {name}!"

def process_greeting(ctx: task.ActivityContext, greeting: str) -> str:
    """Second activity that processes the greeting."""
    return f"{greeting} How are you today?"

def finalize_response(ctx: task.ActivityContext, response: str) -> str:
    """Third activity that finalizes the response."""
    return f"{response} I hope you're doing well!"

Contrairement aux orchestrateurs, les activités peuvent effectuer des opérations d’E/S telles que les appels HTTP, les requêtes de base de données et l’accès aux fichiers. L’entrée est transmise directement en tant que paramètre.

Cet exemple s’affiche pour .NET, JavaScript, Java et Python.

Les activités dans Java sont définies à l’aide de TaskActivityFactory :

import com.microsoft.durabletask.*;

.addActivity(new TaskActivityFactory() {
    @Override
    public String getName() { return "Reverse"; }

    @Override
    public TaskActivity create() {
        return ctx -> {
            String input = ctx.getInput(String.class);
            StringBuilder builder = new StringBuilder(input);
            builder.reverse();
            return builder.toString();
        };
    }
})
.addActivity(new TaskActivityFactory() {
    @Override
    public String getName() { return "Capitalize"; }

    @Override
    public TaskActivity create() {
        return ctx -> ctx.getInput(String.class).toUpperCase();
    }
})
.addActivity(new TaskActivityFactory() {
    @Override
    public String getName() { return "ReplaceWhitespace"; }

    @Override
    public TaskActivity create() {
        return ctx -> {
            String input = ctx.getInput(String.class);
            return input.trim().replaceAll("\\s", "-");
        };
    }
})

Inscrivez chaque activité auprès du constructeur de tâches à l’aide de addActivity. Les activités peuvent effectuer des opérations d’E/S et retourner les résultats à l’orchestrateur.

public static class HttpStart
{
    [FunctionName("HttpStart")]
    public static async Task<HttpResponseMessage> Run(
        [HttpTrigger(AuthorizationLevel.Function, methods: "post", Route = "orchestrators/{functionName}")] HttpRequestMessage req,
        [DurableClient] IDurableClient starter,
        string functionName,
        ILogger log)
    {
        // Function input comes from the request content.
        object eventData = await req.Content.ReadAsAsync<object>();
        string instanceId = await starter.StartNewAsync(functionName, eventData);

        log.LogInformation($"Started orchestration with ID = '{instanceId}'.");

        return starter.CreateCheckStatusResponse(req, instanceId);
    }
}

Pour interagir avec des orchestrateurs, ajoutez une liaison d’entrée DurableClient . Utilisez le client pour démarrer une orchestration et retourner une réponse HTTP qui inclut des URL pour vérifier l’état de la nouvelle orchestration.

HttpStart/function.json

{
  "scriptFile": "__init__.py",
  "bindings": [
    {
      "authLevel": "anonymous",
      "name": "req",
      "type": "httpTrigger",
      "direction": "in",
      "route": "orchestrators/{functionName}",
      "methods": [
        "post",
        "get"
      ]
    },
    {
      "name": "$return",
      "type": "http",
      "direction": "out"
    },
    {
      "name": "starter",
      "type": "durableClient",
      "direction": "in"
    }
  ]
}

Pour interagir avec des orchestrateurs, la fonction doit inclure une liaison d’entrée durableClient.

HttpStart/__init__.py

import logging

import azure.functions as func
import azure.durable_functions as df


async def main(req: func.HttpRequest, starter: str) -> func.HttpResponse:
    client = df.DurableOrchestrationClient(starter)
    instance_id = await client.start_new(req.route_params["functionName"], None, None)

    logging.info(f"Started orchestration with ID = '{instance_id}'.")

    return client.create_check_status_response(req, instance_id)

Utilisez le constructeur DurableOrchestrationClient pour créer un client Durable Functions. Utilisez le client pour démarrer une orchestration et retourner une réponse HTTP qui inclut des URL pour vérifier l’état de la nouvelle orchestration.

Exemple PowerShell bientôt disponible.

Java échantillon sera bientôt disponible.

using Microsoft.DurableTask.Client;

// Create the client
var client = DurableTaskClientBuilder.UseDurableTaskScheduler(connectionString).Build();

// Schedule a new orchestration instance
string instanceId = await client.ScheduleNewOrchestrationInstanceAsync(
    nameof(GreetingOrchestration),
    input: "World");

Console.WriteLine($"Started orchestration with ID: {instanceId}");

// Wait for the orchestration to complete
OrchestrationMetadata result = await client.WaitForInstanceCompletionAsync(
    instanceId,
    getInputsAndOutputs: true);

Console.WriteLine($"Orchestration completed with result: {result.ReadOutputAs<string>()}");

Créez le DurableTaskClient à l’aide d’une chaîne de connexion au Planificateur de tâches durable. Permet ScheduleNewOrchestrationInstanceAsync de démarrer une orchestration et WaitForInstanceCompletionAsync d’attendre la fin.

import {
  DurableTaskAzureManagedClientBuilder,
} from "@microsoft/durabletask-js-azuremanaged";

const connectionString =
  process.env.DURABLE_TASK_SCHEDULER_CONNECTION_STRING ||
  "Endpoint=http://localhost:8080;Authentication=None;TaskHub=default";

const client = new DurableTaskAzureManagedClientBuilder()
  .connectionString(connectionString)
  .build();

// Schedule a new orchestration instance
const instanceId = await client.scheduleNewOrchestration(greetingOrchestrator, "World");
console.log(`Started orchestration with ID: ${instanceId}`);

// Wait for the orchestration to complete
const state = await client.waitForOrchestrationCompletion(instanceId, true, 30);
console.log(`Orchestration completed with result: ${state?.serializedOutput}`);

Créez le DurableTaskAzureManagedClientBuilder à l’aide d’une chaîne de connexion au Planificateur de tâches durable. Permet scheduleNewOrchestration de démarrer une orchestration et waitForOrchestrationCompletion d’attendre la fin.

from durabletask.azuremanaged.client import DurableTaskSchedulerClient

# Create the client
client = DurableTaskSchedulerClient(
    host_address=endpoint,
    secure_channel=endpoint != "http://localhost:8080",
    taskhub=taskhub_name,
    token_credential=credential
)

# Schedule a new orchestration instance
instance_id = client.schedule_new_orchestration(
    function_chaining_orchestrator,
    input="World"
)

print(f"Started orchestration with ID: {instance_id}")

# Wait for the orchestration to complete
result = client.wait_for_orchestration_completion(instance_id, timeout=60)

if result and result.runtime_status == OrchestrationStatus.COMPLETED:
    print(f"Orchestration completed with result: {result.serialized_output}")

Le DurableTaskSchedulerClient se connecte au Planificateur de tâches durable. Permet schedule_new_orchestration de démarrer une orchestration et wait_for_orchestration_completion d’attendre la fin.

Cet exemple s’affiche pour .NET, JavaScript, Java et Python.

import com.microsoft.durabletask.*;
import com.microsoft.durabletask.azuremanaged.DurableTaskSchedulerClientExtensions;

// Create the client
DurableTaskClient client = DurableTaskSchedulerClientExtensions
    .createClientBuilder(connectionString)
    .build();

// Schedule a new orchestration instance
String instanceId = client.scheduleNewOrchestrationInstance(
    "ActivityChaining",
    new NewOrchestrationInstanceOptions().setInput("Hello, world!"));

System.out.println("Started orchestration with ID: " + instanceId);

// Wait for the orchestration to complete
OrchestrationMetadata result = client.waitForInstanceCompletion(
    instanceId,
    Duration.ofSeconds(30),
    true);

System.out.println("Orchestration completed with result: " + result.readOutputAs(String.class));

Créez le DurableTaskClient à l’aide d’un connection string. Permet scheduleNewOrchestrationInstance de démarrer une orchestration et waitForInstanceCompletion d’attendre la fin.