Привязки для Durable Functions (Azure Functions)

Обе версии модели программирования Python для функций Azure поддерживаются устойчивыми функциями. Так как Python v2 является рекомендуемой, примеры в этой статье представлены исключительно в этой версии.

Необходимые условия

• пакет SDK Durable Functions, который является пакетом Индекса пакетов Python (PyPI) azure-functions-durable, версии 1.2.2 или более поздней
• Пакет расширений версии 4.x (или более поздняя версия), который устанавливается в файле проекта host.json

Вы можете предоставить отзывы и предложения в пакете SDK Durable Functions для репозитория Python.

При разработке функций в .NET используйте атрибут OrchestrationTriggerAttribute .NET для настройки триггера оркестрации.

Для Java вы используете аннотацию @DurableOrchestrationTrigger для настройки триггера оркестрации.

При использовании версии 4 модели программирования Node.js для разработки функций вы импортируете app объект из @azure/functions npm модуля. Затем вы вызываете app.orchestration метод API устойчивых функций непосредственно в коде функции. Этот метод регистрирует функцию оркестратора с помощью платформы Durable Functions.

При записи функций оркестратора необходимо определить триггер оркестрации с помощью следующего объекта JSON в массиве bindings файла function.json :

{
    "name": "<name-of-input-parameter-in-function-signature>",
    "orchestration": "<optional-name-of-orchestration>",
    "type": "orchestrationTrigger",
    "direction": "in"
}

Значением orchestration является имя оркестрации, которую клиенты должны использовать при запуске новых экземпляров функции оркестратора. Это свойство является необязательным. Если он не указан, используется имя функции.

При использовании модели программирования Python версии 2 можно определить триггер оркестрации с помощью декоратора orchestration_trigger непосредственно в коде функции Python.

В модели версии 2 вы получаете доступ к триггерам и привязкам в рамках Durable Functions из экземпляра DFApp. Этот подкласс FunctionApp можно использовать для экспорта декораторов, специфических для Durable Functions.

Атрибут, используемый для определения триггера, зависит от того, выполняете ли функции C# в том же процессе, что и процесс хоста функций или в изолированном рабочем процессе.

[FunctionName("HelloWorld")]
public static string RunOrchestrator([OrchestrationTrigger] IDurableOrchestrationContext context)
{
    string name = context.GetInput<string>();
    return $"Hello {name}!";
}

[Function("HelloWorld")]
public static string RunOrchestrator([OrchestrationTrigger] TaskOrchestrationContext context, string name)
{
    return $"Hello {name}!";
}

const { app } = require('@azure/functions');
const df = require('durable-functions');

df.app.orchestration('helloOrchestrator', function* (context) {
    const name = context.df.getInput();
    return `Hello ${name}`;
});

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

myApp = df.DFApp(http_auth_level=func.AuthLevel.ANONYMOUS)

@myApp.orchestration_trigger(context_name="context")
def my_orchestrator(context):
    result = yield context.call_activity("Hello", "Tokyo")
    return result

param($Context)

$InputData = $Context.Input
$InputData

@FunctionName("HelloWorldOrchestration")
public String helloWorldOrchestration(
        @DurableOrchestrationTrigger(name = "ctx") TaskOrchestrationContext ctx) {
    return String.format("Hello %s!", ctx.getInput(String.class));
}

Большинство функций оркестратора вызывают функции действия. Следующий код содержит пример Hello World, демонстрирующий вызов функции действия:

const { app } = require('@azure/functions');
const df = require('durable-functions');

const activityName = 'hello';

df.app.orchestration('helloOrchestrator', function* (context) {
    const name = context.df.getInput();
    const result = yield context.df.callActivity(activityName, name);
    return result;
});

@FunctionName("HelloWorld")
public String helloWorldOrchestration(
        @DurableOrchestrationTrigger(name = "ctx") TaskOrchestrationContext ctx) {
    String input = ctx.getInput(String.class);
    String result = ctx.callActivity("SayHello", input, String.class).await();
    return result;
}

Для настройки триггера действия используется атрибут ActivityTriggerAttribute .NET.

С помощью заметки @DurableActivityTrigger можно настроить триггер действия.

Чтобы зарегистрировать функцию действия, импортируйте app объект из @azure/functions npm модуля. Затем вы вызываете app.activity метод API устойчивых функций непосредственно в коде функции.

Чтобы определить триггер действия, используйте следующий объект JSON в массиве bindingsfunction.json:

{
    "name": "<name-of-input-parameter-in-function-signature>",
    "activity": "<optional-name-of-activity>",
    "type": "activityTrigger",
    "direction": "in"
}

Значением activity является имя действия. Это значение — имя, используемое функциями оркестратора для вызова этой функции действия. Это свойство является необязательным. Если он не указан, используется имя функции.

Триггер действия можно определить с помощью декоратора activity_trigger непосредственно в коде функции Python.

const { app } = require('@azure/functions');
const df = require('durable-functions');
const activityName = 'hello';
df.app.activity(activityName, {
    handler: (input) => {
        return `Hello, ${input}`;
    },
});

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

myApp = df.DFApp(http_auth_level=func.AuthLevel.ANONYMOUS)

@myApp.activity_trigger(input_name="myInput")
def my_activity(myInput: str):
    return "Hello " + myInput

param($name)

"Hello $name!"

@FunctionName("SayHello")
public String sayHello(@DurableActivityTrigger(name = "name") String name) {
    return String.format("Hello %s!", name);
}

Например, функция действия может получать входные данные из функции оркестратора. Затем функция действия может отправлять эти входные данные в виде сообщения в Azure Event Hubs.

const { app } = require('@azure/functions');
const df = require('durable-functions');

df.app.orchestration('helloOrchestrator', function* (context) {
    const input = context.df.getInput();
    yield context.df.callActivity('sendToEventHub', input);
    return `Message sent: ${input}`;
});

const { EventHubProducerClient } = require("@azure/event-hubs");
const connectionString = process.env.EVENT_HUB_CONNECTION_STRING;
const eventHubName = process.env.EVENT_HUB_NAME;

df.app.activity("sendToEventHub", {
    handler: async (message, context) => {
        const producer = new EventHubProducerClient(connectionString, eventHubName);
        try {
            const batch = await producer.createBatch();
            batch.tryAdd({ body: message });
            await producer.sendBatch(batch);
            context.log(`Message sent to Event Hubs: ${message}`);
        } catch (err) {
            context.log.error("Failed to send message to Event Hubs:", err);
            throw err;
        } finally {
            await producer.close();
        }
    },
});

app.storageQueue('helloQueueStart', {
    queueName: 'start-orchestration',
    extraInputs: [df.input.durableClient()],
    handler: async (message, context) => {
        const client = df.getClient(context);
        const orchestratorName = message.orchestratorName || 'helloOrchestrator';
        const input = message.input || null;
        const instanceId = await client.startNew(orchestratorName, { input });
        context.log(`Started orchestration with ID = '${instanceId}'`);
    },
});

Вы можете привязаться к клиенту оркестрации с помощью атрибута DurableClientAttribute (OrchestrationClientAttribute в Durable Functions 1.x).

Вы можете привязаться к клиенту оркестрации с помощью аннотации @DurableClientInput.

Чтобы зарегистрировать клиентную функцию, импортируйте app объект из @azure/functions npm модуля. Затем вы вызываете метод API устойчивых функций, который зависит от типа триггера. Например, для триггера HTTP вызывается app.http метод. Для триггера очереди вызывается app.storageQueue метод.

Чтобы определить устойчивый триггер клиента, используйте следующий объект JSON в массиве bindingsfunction.json:

{
    "name": "<name-of-input-parameter-in-function-signature>",
    "taskHub": "<optional-name-of-task-hub>",
    "connectionName": "<optional-name-of-connection-string-app-setting>",
    "type": "orchestrationClient",
    "direction": "in"
}

• Это taskHub свойство используется, если несколько приложений-функций используют одну и ту же учетную запись хранения, но их необходимо изолировать друг от друга. Если это свойство не указано, используется значение по умолчанию из host.json . Это значение должно соответствовать значению, используемому целевыми функциями оркестратора.
• Значение connectionName — это имя параметра приложения, содержащего строку подключения учетной записи хранения. Учетная запись хранения, представленная этой строкой подключения, должна совпадать с той, которую используют целевые функции оркестратора. Если это свойство не указано, используется строка подключения учетной записи хранения по умолчанию для функционального приложения.

Вы можете определить устойчивый триггер клиента с помощью декоратора durable_client_input непосредственно в коде функции Python.

Обычно выполняется привязка к реализации IDurableClient (DurableOrchestrationClient в Durable Functions версии 1.x), которая обеспечивает полный доступ ко всем API клиента оркестрации, поддерживаемым Durable Functions.

Обычно вы привязываетесь к классу DurableClientContext.

Чтобы получить доступ к объекту клиента, необходимо использовать пакет SDK для конкретного языка.

const { app } = require('@azure/functions');
const df = require('durable-functions');

app.storageQueue('helloQueueStart', {
    queueName: 'start-orchestration',
    extraInputs: [df.input.durableClient()],
    handler: async (message, context) => {
        const client = df.getClient(context);
        const orchestratorName = message.orchestratorName || 'helloOrchestrator';
        const input = message.input || null;
        const instanceId = await client.startNew(orchestratorName, { input });
        context.log(`Started orchestration with ID = '${instanceId}' from queue message.`);
    },
});

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

myApp = df.DFApp(http_auth_level=func.AuthLevel.ANONYMOUS)

@myApp.queue_trigger(
    arg_name="msg",
    queue_name="start-orchestration",
    connection="AzureWebJobsStorage"
)
@myApp.durable_client_input(client_name="client")
async def client_function(msg: func.QueueMessage, client: df.DurableOrchestrationClient):
    input_data = msg.get_body().decode("utf-8")
    await client.start_new("my_orchestrator", None, input_data)
    return None

function.json

{
  "bindings": [
    {
      "name": "InputData",
      "type": "queueTrigger",
      "queueName": "durable-function-trigger",
      "direction": "in"
    },
    {
      "name": "starter",
      "type": "durableClient",
      "direction": "in"
    }
  ]
}

run.ps1

param([string]$InputData, $TriggerMetadata)

$InstanceId = Start-DurableOrchestration -FunctionName 'HelloWorld' -Input $InputData

@FunctionName("QueueStart")
public void queueStart(
        @QueueTrigger(name = "input", queueName = "durable-function-trigger", connection = "Storage") String input,
        @DurableClientInput(name = "durableContext") DurableClientContext durableContext) {
    // Orchestration input comes from the queue message content.
    durableContext.getClient().scheduleNewOrchestrationInstance("HelloWorld", input);
}

Для настройки триггера сущности используется атрибут EntityTriggerAttribute .NET.

Чтобы зарегистрировать триггер сущности, импортируйте объект app из модуля @azure/functions npm. Затем вы вызываете app.entity метод API устойчивых функций непосредственно в коде функции.

const df = require('durable-functions');
df.app.entity('counter', (context) => {
    const currentValue = context.df.getState(() => 0);
    switch (context.df.operationName) {
        case 'add':
            context.df.setState(currentValue + context.df.getInput());
            break;
        case 'reset':
            context.df.setState(0);
            break;
        case 'get':
            context.df.return(currentValue);
            break;
    }
});

Триггер сущности можно определить с помощью декоратора entity_trigger непосредственно в коде функции Python.

Вы можете привязаться к клиенту сущности с помощью атрибута DurableClientAttribute .NET в функциях библиотеки классов .NET.

Вместо регистрации клиента сущности, вы используете signalEntity или callEntity, чтобы вызвать метод триггера сущности из любой зарегистрированной функции.

• Из функции, триггеруемой очередью, можно использовать client.signalEntity:

  const { app } = require('@azure/functions');
  const df = require('durable-functions');
  app.storageQueue('helloQueueStart', {
      queueName: 'start-orchestration',
      extraInputs: [df.input.durableClient()],
      handler: async (message, context) => {
          const client = df.getClient(context);
          const entityId = new df.EntityId('counter', 'myCounter');
          await client.signalEntity(entityId, 'add', 5);
      },
  });
  

• Из функции оркестратора можно использовать context.df.callEntity:

  const { app } = require('@azure/functions');
  const df = require('durable-functions');
  df.app.orchestration('entityCaller', function* (context) {
      const entityId = new df.EntityId('counter', 'myCounter');
      yield context.df.callEntity(entityId, 'add', 5);
      yield context.df.callEntity(entityId, 'add', 5);
      const result = yield context.df.callEntity(entityId, 'get');
      return result;
  });
  

Клиентский объект сущности можно определить с помощью декоратора durable_client_input непосредственно в коде функции Python.