diff --git a/samples/durable-functions/dotnet/WorkItemFiltering/Functions.cs b/samples/durable-functions/dotnet/WorkItemFiltering/Functions.cs new file mode 100644 index 0000000..71eae0d --- /dev/null +++ b/samples/durable-functions/dotnet/WorkItemFiltering/Functions.cs @@ -0,0 +1,162 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. + +using System.Net; +using Microsoft.Azure.Functions.Worker; +using Microsoft.Azure.Functions.Worker.Http; +using Microsoft.DurableTask; +using Microsoft.DurableTask.Client; +using Microsoft.DurableTask.Entities; +using Microsoft.Extensions.Logging; + +namespace WorkItemFiltering; + +// ============================================================================= +// Orchestrations +// ============================================================================= + +/// +/// A simple orchestration that calls an activity and returns the result. +/// With work item filtering enabled, DTS will only dispatch this orchestration +/// to workers that have it registered. +/// +public static class GreetingOrchestration +{ + [Function(nameof(GreetingOrchestration))] + public static async Task Run([OrchestrationTrigger] TaskOrchestrationContext ctx) + { + ctx.CreateReplaySafeLogger(nameof(GreetingOrchestration)).LogInformation("GreetingOrchestration started"); + return await ctx.CallActivityAsync(nameof(SayHello), "World"); + } + + [Function(nameof(GreetingOrchestration) + "_Start")] + public static async Task Start( + [HttpTrigger(AuthorizationLevel.Anonymous, "post", Route = "orchestrators/greeting")] HttpRequestData req, + [DurableClient] DurableTaskClient client) + { + string instanceId = await client.ScheduleNewOrchestrationInstanceAsync(nameof(GreetingOrchestration)); + return client.CreateCheckStatusResponse(req, instanceId); + } +} + +/// +/// A fan-out/fan-in orchestration that calls the same activity in parallel. +/// Demonstrates that activity work items are also filtered. +/// +public static class FanOutOrchestration +{ + [Function(nameof(FanOutOrchestration))] + public static async Task Run([OrchestrationTrigger] TaskOrchestrationContext ctx) + { + ctx.CreateReplaySafeLogger(nameof(FanOutOrchestration)).LogInformation("FanOutOrchestration: fanning out to 3 activities"); + return await Task.WhenAll( + ctx.CallActivityAsync(nameof(SayHello), "Tokyo"), + ctx.CallActivityAsync(nameof(SayHello), "London"), + ctx.CallActivityAsync(nameof(SayHello), "Seattle")); + } + + [Function(nameof(FanOutOrchestration) + "_Start")] + public static async Task Start( + [HttpTrigger(AuthorizationLevel.Anonymous, "post", Route = "orchestrators/fanout")] HttpRequestData req, + [DurableClient] DurableTaskClient client) + { + string instanceId = await client.ScheduleNewOrchestrationInstanceAsync(nameof(FanOutOrchestration)); + return client.CreateCheckStatusResponse(req, instanceId); + } +} + +/// +/// A parent orchestration that calls a child orchestration. +/// Sub-orchestration dispatch is also governed by work item filters. +/// +public static class ParentOrchestration +{ + [Function(nameof(ParentOrchestration))] + public static async Task Run([OrchestrationTrigger] TaskOrchestrationContext ctx) + { + ctx.CreateReplaySafeLogger(nameof(ParentOrchestration)).LogInformation("Calling sub-orchestration"); + string result = await ctx.CallSubOrchestratorAsync(nameof(GreetingOrchestration)); + return $"Parent received: {result}"; + } + + [Function(nameof(ParentOrchestration) + "_Start")] + public static async Task Start( + [HttpTrigger(AuthorizationLevel.Anonymous, "post", Route = "orchestrators/parent")] HttpRequestData req, + [DurableClient] DurableTaskClient client) + { + string instanceId = await client.ScheduleNewOrchestrationInstanceAsync(nameof(ParentOrchestration)); + return client.CreateCheckStatusResponse(req, instanceId); + } +} + +/// +/// An orchestration that interacts with a durable entity. +/// Entity work items are also filtered. +/// +public static class CounterOrchestration +{ + [Function(nameof(CounterOrchestration))] + public static async Task Run([OrchestrationTrigger] TaskOrchestrationContext ctx) + { + var logger = ctx.CreateReplaySafeLogger(nameof(CounterOrchestration)); + var entityId = new EntityInstanceId(nameof(CounterEntity), "sample-counter"); + + await ctx.Entities.CallEntityAsync(entityId, "Add", 10); + await ctx.Entities.CallEntityAsync(entityId, "Add", 20); + int value = await ctx.Entities.CallEntityAsync(entityId, "Get"); + + logger.LogInformation("Counter value = {Value}", value); + return value; + } + + [Function(nameof(CounterOrchestration) + "_Start")] + public static async Task Start( + [HttpTrigger(AuthorizationLevel.Anonymous, "post", Route = "orchestrators/counter")] HttpRequestData req, + [DurableClient] DurableTaskClient client) + { + string instanceId = await client.ScheduleNewOrchestrationInstanceAsync(nameof(CounterOrchestration)); + return client.CreateCheckStatusResponse(req, instanceId); + } +} + +// ============================================================================= +// Activities +// ============================================================================= + +public static class SayHello +{ + [Function(nameof(SayHello))] + public static string Run([ActivityTrigger] string name) => $"Hello, {name}!"; +} + +// ============================================================================= +// Entities +// ============================================================================= + +public class CounterEntity : TaskEntity +{ + public void Add(int amount) => this.State += amount; + public void Reset() => this.State = 0; + public int Get() => this.State; + + [Function(nameof(CounterEntity))] + public static Task Dispatch([EntityTrigger] TaskEntityDispatcher dispatcher) + => dispatcher.DispatchAsync(); +} + +// ============================================================================= +// Generic starter (for cross-app filter isolation tests) +// ============================================================================= + +public static class GenericStarter +{ + [Function("StartOrchestration")] + public static async Task Start( + [HttpTrigger(AuthorizationLevel.Anonymous, "post", Route = "start/{name}")] HttpRequestData req, + [DurableClient] DurableTaskClient client, + string name) + { + string instanceId = await client.ScheduleNewOrchestrationInstanceAsync(name); + return client.CreateCheckStatusResponse(req, instanceId); + } +} diff --git a/samples/durable-functions/dotnet/WorkItemFiltering/Program.cs b/samples/durable-functions/dotnet/WorkItemFiltering/Program.cs new file mode 100644 index 0000000..46bca8a --- /dev/null +++ b/samples/durable-functions/dotnet/WorkItemFiltering/Program.cs @@ -0,0 +1,5 @@ +using Microsoft.Azure.Functions.Worker.Builder; +using Microsoft.Extensions.Hosting; + +FunctionsApplicationBuilder builder = FunctionsApplication.CreateBuilder(args); +builder.Build().Run(); diff --git a/samples/durable-functions/dotnet/WorkItemFiltering/README.md b/samples/durable-functions/dotnet/WorkItemFiltering/README.md new file mode 100644 index 0000000..1a8779c --- /dev/null +++ b/samples/durable-functions/dotnet/WorkItemFiltering/README.md @@ -0,0 +1,130 @@ +# Work Item Filtering with Durable Functions (.NET) + +.NET | Durable Functions + +## Description + +Demonstrates the **work item filtering** feature for Durable Functions with the Durable Task Scheduler (DTS) backend. When multiple Function apps share the same DTS task hub, work item filtering ensures each app only receives work items for the functions it has registered — preventing dispatch failures. + +This sample includes orchestrations, activities, entities, sub-orchestrations, and fan-out/fan-in patterns — all governed by work item filters. + +## Prerequisites + +1. [.NET 8 SDK](https://dotnet.microsoft.com/download/dotnet/8.0) +2. [Docker](https://www.docker.com/products/docker-desktop/) (for running the emulator and Azurite) +3. [Azure Functions Core Tools v4](https://learn.microsoft.com/azure/azure-functions/functions-run-local) + +## Quick Run + +1. Start the Durable Task Scheduler emulator: + ```bash + docker run --name dtsemulator -d -p 8080:8080 -p 8082:8082 mcr.microsoft.com/dts/dts-emulator:latest + ``` + +2. Start Azurite (Azure Storage emulator): + ```bash + docker run --name azurite -d -p 10000:10000 -p 10001:10001 -p 10002:10002 mcr.microsoft.com/azure-storage/azurite + ``` + +3. Start the Function app: + ```bash + func start + ``` + +4. Trigger an orchestration: + ```bash + curl -X POST http://localhost:7071/api/orchestrators/greeting + ``` + +5. Test filter isolation — schedule an orchestration this app does NOT have: + ```bash + curl -X POST http://localhost:7071/api/start/SomeOtherOrchestration + ``` + Check the status — it should stay `Pending` because no worker has `SomeOtherOrchestration` in its filter. Without filtering, this would fail with *"function doesn't exist"*. + +## Expected Output + +``` +# Matching orchestration → Completed +{"name":"GreetingOrchestration","runtimeStatus":"Completed","output":"Hello, World!"} + +# Unknown orchestration → stays Pending (filter isolation working) +{"name":"SomeOtherOrchestration","runtimeStatus":"Pending"} +``` + +## How It Works + +The key configuration in [`host.json`](host.json): + +```json +{ + "extensions": { + "durableTask": { + "storageProvider": { + "type": "azureManaged", + "connectionStringName": "DURABLE_TASK_SCHEDULER_CONNECTION_STRING", + "workItemFilteringEnabled": true + } + } + } +} +``` + +When `workItemFilteringEnabled` is `true`: +1. The Durable Functions extension discovers registered orchestrators, activities, and entities during function indexing +2. These names are sent to DTS as `WorkItemFilters` on the `GetWorkItems` gRPC stream +3. DTS only dispatches work items that match the worker's registered functions +4. Unmatched work items stay in the DTS queue until a matching worker connects + +No code changes are needed — filtering is automatic based on the functions registered in the app. + +## Registered Functions + +| Type | Function | Description | +|---------------|---------------------------|-------------------------------------| +| Orchestration | `GreetingOrchestration` | Simple activity call | +| Orchestration | `FanOutOrchestration` | Parallel fan-out to 3 activities | +| Orchestration | `ParentOrchestration` | Calls GreetingOrchestration as sub | +| Orchestration | `CounterOrchestration` | Interacts with CounterEntity | +| Activity | `SayHello` | Returns a greeting string | +| Entity | `CounterEntity` | Counter with Add/Reset/Get | + +## Multi-App Scenario + +To see filter isolation in action across two apps: + +1. Create a second Function app with **different** orchestrations/activities +2. Point both apps to the **same** DTS task hub +3. Enable `workItemFilteringEnabled: true` in both +4. Schedule orchestrations — each app only processes its own functions + +## Viewing in the Dashboard + +- **Emulator:** Navigate to http://localhost:8082 → select the "default" task hub +- **Azure:** Navigate to your Scheduler resource in the Azure Portal → Task Hub → Dashboard URL + +## Using a Deployed Scheduler (Azure) + +To use a Durable Task Scheduler in Azure instead of the emulator: + +1. Update `local.settings.json`: + ```json + { + "Values": { + "DURABLE_TASK_SCHEDULER_CONNECTION_STRING": "Endpoint=https://.durabletask.io;Authentication=ManagedIdentity" + } + } + ``` + +2. Run the sample using the same commands above. + +## Related Samples + +- [WorkItemFilteringSplitActivities](../../../scenarios/WorkItemFilteringSplitActivities/) — Multi-worker scenario using Durable Task SDK +- [Fan-out/Fan-in (Python)](../../python/fan-out-fan-in/) — Fan-out pattern in Python +- [HelloCities (.NET)](../HelloCities/) — Basic Durable Functions quickstart + +## Learn More + +- [Durable Task Scheduler documentation](https://learn.microsoft.com/azure/azure-functions/durable/durable-functions-task-hubs) +- [Durable Functions patterns](https://learn.microsoft.com/azure/azure-functions/durable/durable-functions-overview) diff --git a/samples/durable-functions/dotnet/WorkItemFiltering/WorkItemFiltering.csproj b/samples/durable-functions/dotnet/WorkItemFiltering/WorkItemFiltering.csproj new file mode 100644 index 0000000..cc3bb4c --- /dev/null +++ b/samples/durable-functions/dotnet/WorkItemFiltering/WorkItemFiltering.csproj @@ -0,0 +1,16 @@ + + + net8.0 + v4 + Exe + enable + enable + + + + + + + + + diff --git a/samples/durable-functions/dotnet/WorkItemFiltering/host.json b/samples/durable-functions/dotnet/WorkItemFiltering/host.json new file mode 100644 index 0000000..da06fd0 --- /dev/null +++ b/samples/durable-functions/dotnet/WorkItemFiltering/host.json @@ -0,0 +1,19 @@ +{ + "version": "2.0", + "logging": { + "logLevel": { + "DurableTask.AzureStorage": "Warning", + "DurableTask.Core": "Warning" + } + }, + "extensions": { + "durableTask": { + "hubName": "default", + "storageProvider": { + "type": "azureManaged", + "connectionStringName": "DURABLE_TASK_SCHEDULER_CONNECTION_STRING", + "workItemFilteringEnabled": true + } + } + } +} diff --git a/samples/durable-functions/dotnet/WorkItemFiltering/test.http b/samples/durable-functions/dotnet/WorkItemFiltering/test.http new file mode 100644 index 0000000..84acc9a --- /dev/null +++ b/samples/durable-functions/dotnet/WorkItemFiltering/test.http @@ -0,0 +1,92 @@ +# ============================================================================= +# Work Item Filtering — Durable Functions (.NET) Demo +# ============================================================================= +# +# Prerequisites: +# 1. DTS emulator: docker run -d -p 8080:8080 -p 8082:8082 mcr.microsoft.com/dts/dts-emulator:latest +# 2. Azurite: docker run -d -p 10000:10000 -p 10001:10001 -p 10002:10002 mcr.microsoft.com/azure-storage/azurite +# 3. App running: func start --port 7071 +# +# This app registers: GreetingOrchestration, FanOutOrchestration, +# ParentOrchestration, CounterOrchestration, SayHello, CounterEntity +# +# With workItemFilteringEnabled: true, DTS only dispatches matching work. +# Unmatched orchestrations stay Pending — proving filter isolation. +# ============================================================================= + +@baseUrl = http://localhost:7071/api + + +# --- TEST 1: Simple orchestration + activity --- +# Expected: Completed with "Hello, World!" + +### Start GreetingOrchestration +# @name greeting +POST {{baseUrl}}/orchestrators/greeting + +### Check status +GET {{baseUrl}}/instances/{{greeting.response.body.id}} + + +# --- TEST 2: Fan-out/fan-in (parallel activities) --- +# Expected: Completed with ["Hello, Tokyo!", "Hello, London!", "Hello, Seattle!"] + +### Start FanOutOrchestration +# @name fanout +POST {{baseUrl}}/orchestrators/fanout + +### Check status +GET {{baseUrl}}/instances/{{fanout.response.body.id}} + + +# --- TEST 3: Sub-orchestration --- +# Expected: Completed with "Parent received: Hello, World!" + +### Start ParentOrchestration +# @name parent +POST {{baseUrl}}/orchestrators/parent + +### Check status +GET {{baseUrl}}/instances/{{parent.response.body.id}} + + +# --- TEST 4: Entity interaction --- +# Expected: Completed with 30 (10 + 20) + +### Start CounterOrchestration +# @name counter +POST {{baseUrl}}/orchestrators/counter + +### Check status +GET {{baseUrl}}/instances/{{counter.response.body.id}} + + +# --- TEST 5: Filter isolation — unknown orchestration --- +# Expected: Stays Pending (no worker has this function registered) + +### Start an orchestration this app does NOT have +# @name unknown +POST {{baseUrl}}/start/SomeOtherOrchestration + +### Check status — should be Pending +GET {{baseUrl}}/instances/{{unknown.response.body.id}} + +### Check again after 15 seconds — still Pending +GET {{baseUrl}}/instances/{{unknown.response.body.id}} + + +# --- TEST 6: View all instances --- + +### List all orchestration instances +GET http://localhost:7071/runtime/webhooks/durabletask/instances + + +# --- EXPECTED RESULTS --- +# +# | Test | Orchestration | Status | Why | +# |------|--------------------------|-----------|-----------------------------| +# | 1 | GreetingOrchestration | Completed | Matches filter | +# | 2 | FanOutOrchestration | Completed | Activities also filtered | +# | 3 | ParentOrchestration | Completed | Sub-orch also filtered | +# | 4 | CounterOrchestration | Completed | Entities also filtered | +# | 5 | SomeOtherOrchestration | Pending | No matching filter → held | diff --git a/samples/durable-functions/python/work-item-filtering/README.md b/samples/durable-functions/python/work-item-filtering/README.md new file mode 100644 index 0000000..10cef76 --- /dev/null +++ b/samples/durable-functions/python/work-item-filtering/README.md @@ -0,0 +1,122 @@ +# Work Item Filtering with Durable Functions (Python) + +Python | Durable Functions + +## Description + +Demonstrates the **work item filtering** feature for Durable Functions with the Durable Task Scheduler (DTS) backend. When multiple Function apps share the same DTS task hub, work item filtering ensures each app only receives work items for the functions it has registered — preventing dispatch failures. + +This sample includes orchestrations, activities, entities, sub-orchestrations, and fan-out/fan-in patterns — all governed by work item filters. + +## Prerequisites + +1. [Python 3.9+](https://www.python.org/downloads/) +2. [Docker](https://www.docker.com/products/docker-desktop/) (for running the emulator and Azurite) +3. [Azure Functions Core Tools v4](https://learn.microsoft.com/azure/azure-functions/functions-run-local) + +## Quick Run + +1. Start the Durable Task Scheduler emulator: + ```bash + docker run --name dtsemulator -d -p 8080:8080 -p 8082:8082 mcr.microsoft.com/dts/dts-emulator:latest + ``` + +2. Start Azurite (Azure Storage emulator): + ```bash + docker run --name azurite -d -p 10000:10000 -p 10001:10001 -p 10002:10002 mcr.microsoft.com/azure-storage/azurite + ``` + +3. Set up the Python environment and start the Function app: + ```bash + python -m venv .venv + source .venv/bin/activate # Linux/macOS + # .venv\Scripts\activate # Windows + pip install -r requirements.txt + func start + ``` + +4. Trigger an orchestration: + ```bash + curl -X POST http://localhost:7071/api/orchestrators/greeting + ``` + +5. Test filter isolation — schedule an orchestration this app does NOT have: + ```bash + curl -X POST http://localhost:7071/api/start/SomeOtherOrchestration + ``` + Check the status — it should stay `Pending` because no worker has `SomeOtherOrchestration` in its filter. + +## Expected Output + +``` +# Matching orchestration → Completed +{"runtimeStatus": "Completed", "output": "Hello, World!"} + +# Unknown orchestration → stays Pending (filter isolation working) +{"runtimeStatus": "Pending"} +``` + +## How It Works + +The key configuration in [`host.json`](host.json): + +```json +{ + "extensions": { + "durableTask": { + "storageProvider": { + "type": "azureManaged", + "connectionStringName": "DURABLE_TASK_SCHEDULER_CONNECTION_STRING", + "workItemFilteringEnabled": true + } + } + } +} +``` + +When `workItemFilteringEnabled` is `true`: +1. The Durable Functions extension discovers registered orchestrators, activities, and entities during function indexing +2. These names are sent to DTS as `WorkItemFilters` on the `GetWorkItems` gRPC stream +3. DTS only dispatches work items that match the worker's registered functions +4. Unmatched work items stay in the DTS queue until a matching worker connects + +No code changes are needed — filtering is automatic based on the functions registered in the app. This works the same way for all languages (Python, .NET, JavaScript, Java). + +## Registered Functions + +| Type | Function | Description | +|---------------|---------------------------|-------------------------------------| +| Orchestration | `greeting_orchestration` | Simple activity call | +| Orchestration | `fan_out_orchestration` | Parallel fan-out to 3 activities | +| Orchestration | `parent_orchestration` | Calls greeting as sub-orchestration | +| Orchestration | `counter_orchestration` | Interacts with counter_entity | +| Activity | `say_hello` | Returns a greeting string | +| Entity | `counter_entity` | Counter with add/reset/get | + +## Viewing in the Dashboard + +- **Emulator:** Navigate to http://localhost:8082 → select the "default" task hub +- **Azure:** Navigate to your Scheduler resource in the Azure Portal → Task Hub → Dashboard URL + +## Using a Deployed Scheduler (Azure) + +To use a Durable Task Scheduler in Azure instead of the emulator, update `local.settings.json`: + +```json +{ + "Values": { + "DURABLE_TASK_SCHEDULER_CONNECTION_STRING": "Endpoint=https://.durabletask.io;Authentication=ManagedIdentity" + } +} +``` + +## Related Samples + +- [Work Item Filtering (.NET)](../../dotnet/WorkItemFiltering/) — Same feature in .NET +- [WorkItemFilteringSplitActivities](../../../scenarios/WorkItemFilteringSplitActivities/) — Multi-worker scenario using Durable Task SDK +- [Fan-out/Fan-in (Python)](../fan-out-fan-in/) — Fan-out pattern without filtering + +## Learn More + +- [Durable Task Scheduler documentation](https://learn.microsoft.com/azure/azure-functions/durable/durable-functions-task-hubs) +- [Durable Functions for Python](https://learn.microsoft.com/azure/azure-functions/durable/quickstart-python-vscode) diff --git a/samples/durable-functions/python/work-item-filtering/function_app.py b/samples/durable-functions/python/work-item-filtering/function_app.py new file mode 100644 index 0000000..54e79cc --- /dev/null +++ b/samples/durable-functions/python/work-item-filtering/function_app.py @@ -0,0 +1,148 @@ +# Copyright (c) Microsoft Corporation. All rights reserved. +# Licensed under the MIT License. + +""" +Work Item Filtering sample for Durable Functions (Python). + +With workItemFilteringEnabled: true in host.json, this app advertises its +registered orchestrations, activities, and entities to the Durable Task +Scheduler (DTS). DTS then only dispatches matching work items to this worker. + +Orchestrations scheduled for functions NOT registered in this app will stay +in Pending state until a matching worker connects — proving filter isolation. +""" + +import logging +import json +import azure.functions as func +import azure.durable_functions as df + +app = func.FunctionApp(http_auth_level=func.AuthLevel.ANONYMOUS) +bp = df.Blueprint() + + +# ============================================================================= +# Orchestrations +# ============================================================================= + +@bp.orchestration_trigger(context_name="context") +def greeting_orchestration(context: df.DurableOrchestrationContext): + """Simple orchestration that calls an activity.""" + result = yield context.call_activity("say_hello", "World") + return result + + +@bp.orchestration_trigger(context_name="context") +def fan_out_orchestration(context: df.DurableOrchestrationContext): + """Fan-out/fan-in: calls the same activity in parallel with different inputs.""" + cities = ["Tokyo", "London", "Seattle"] + parallel_tasks = [context.call_activity("say_hello", city) for city in cities] + results = yield context.task_all(parallel_tasks) + return results + + +@bp.orchestration_trigger(context_name="context") +def parent_orchestration(context: df.DurableOrchestrationContext): + """Parent orchestration that calls a child orchestration.""" + result = yield context.call_sub_orchestrator("greeting_orchestration") + return f"Parent received: {result}" + + +@bp.orchestration_trigger(context_name="context") +def counter_orchestration(context: df.DurableOrchestrationContext): + """Orchestration that interacts with a durable entity.""" + entity_id = df.EntityId("counter_entity", "sample-counter") + + yield context.call_entity(entity_id, "add", 10) + yield context.call_entity(entity_id, "add", 20) + value = yield context.call_entity(entity_id, "get") + + return value + + +# ============================================================================= +# Activities +# ============================================================================= + +@bp.activity_trigger(input_name="name") +def say_hello(name: str) -> str: + """Simple activity that returns a greeting.""" + logging.info(f"say_hello called with: {name}") + return f"Hello, {name}!" + + +# ============================================================================= +# Entities +# ============================================================================= + +@bp.entity_trigger(context_name="context") +def counter_entity(context: df.DurableEntityContext): + """Simple counter entity with add, reset, and get operations.""" + state = context.get_state(lambda: 0) + operation = context.operation_name + + if operation == "add": + amount = context.get_input() + state += amount + elif operation == "reset": + state = 0 + elif operation == "get": + context.set_result(state) + + context.set_state(state) + + +# ============================================================================= +# HTTP triggers +# ============================================================================= + +@bp.durable_client_input(client_name="client") +@app.route(route="orchestrators/greeting", methods=["POST"]) +async def start_greeting(req: func.HttpRequest, client) -> func.HttpResponse: + """Start the greeting orchestration.""" + client = df.DurableOrchestrationClient(client) + instance_id = await client.start_new("greeting_orchestration") + return client.create_check_status_response(req, instance_id) + + +@bp.durable_client_input(client_name="client") +@app.route(route="orchestrators/fanout", methods=["POST"]) +async def start_fan_out(req: func.HttpRequest, client) -> func.HttpResponse: + """Start the fan-out orchestration.""" + client = df.DurableOrchestrationClient(client) + instance_id = await client.start_new("fan_out_orchestration") + return client.create_check_status_response(req, instance_id) + + +@bp.durable_client_input(client_name="client") +@app.route(route="orchestrators/parent", methods=["POST"]) +async def start_parent(req: func.HttpRequest, client) -> func.HttpResponse: + """Start the parent orchestration.""" + client = df.DurableOrchestrationClient(client) + instance_id = await client.start_new("parent_orchestration") + return client.create_check_status_response(req, instance_id) + + +@bp.durable_client_input(client_name="client") +@app.route(route="orchestrators/counter", methods=["POST"]) +async def start_counter(req: func.HttpRequest, client) -> func.HttpResponse: + """Start the counter orchestration (entity interaction).""" + client = df.DurableOrchestrationClient(client) + instance_id = await client.start_new("counter_orchestration") + return client.create_check_status_response(req, instance_id) + + +@bp.durable_client_input(client_name="client") +@app.route(route="start/{name}", methods=["POST"]) +async def start_any(req: func.HttpRequest, client) -> func.HttpResponse: + """Generic starter — can schedule any orchestration by name. + Useful for testing filter isolation: schedule an orchestration this app + does NOT have and observe it stays Pending. + """ + name = req.route_params.get("name") + client = df.DurableOrchestrationClient(client) + instance_id = await client.start_new(name) + return client.create_check_status_response(req, instance_id) + + +app.register_functions(bp) diff --git a/samples/durable-functions/python/work-item-filtering/host.json b/samples/durable-functions/python/work-item-filtering/host.json new file mode 100644 index 0000000..3bf3ab4 --- /dev/null +++ b/samples/durable-functions/python/work-item-filtering/host.json @@ -0,0 +1,22 @@ +{ + "version": "2.0", + "logging": { + "logLevel": { + "DurableTask.Core": "Warning" + } + }, + "extensions": { + "durableTask": { + "hubName": "default", + "storageProvider": { + "type": "azureManaged", + "connectionStringName": "DURABLE_TASK_SCHEDULER_CONNECTION_STRING", + "workItemFilteringEnabled": true + } + } + }, + "extensionBundle": { + "id": "Microsoft.Azure.Functions.ExtensionBundle", + "version": "[4.*, 5.0.0)" + } +} diff --git a/samples/durable-functions/python/work-item-filtering/requirements.txt b/samples/durable-functions/python/work-item-filtering/requirements.txt new file mode 100644 index 0000000..4d2b03a --- /dev/null +++ b/samples/durable-functions/python/work-item-filtering/requirements.txt @@ -0,0 +1,2 @@ +azure-functions +azure-functions-durable diff --git a/samples/durable-functions/python/work-item-filtering/test.http b/samples/durable-functions/python/work-item-filtering/test.http new file mode 100644 index 0000000..9199b9f --- /dev/null +++ b/samples/durable-functions/python/work-item-filtering/test.http @@ -0,0 +1,80 @@ +# ============================================================================= +# Work Item Filtering — Durable Functions (Python) Demo +# ============================================================================= +# +# Prerequisites: +# 1. DTS emulator: docker run -d -p 8080:8080 -p 8082:8082 mcr.microsoft.com/dts/dts-emulator:latest +# 2. Azurite: docker run -d -p 10000:10000 -p 10001:10001 -p 10002:10002 mcr.microsoft.com/azure-storage/azurite +# 3. App running: python -m venv .venv && .venv/Scripts/activate && pip install -r requirements.txt && func start +# +# Registered functions: greeting_orchestration, fan_out_orchestration, +# parent_orchestration, counter_orchestration, say_hello, counter_entity +# ============================================================================= + +@baseUrl = http://localhost:7071/api + + +# --- TEST 1: Simple orchestration + activity --- +# Expected: Completed with "Hello, World!" + +### Start greeting orchestration +# @name greeting +POST {{baseUrl}}/orchestrators/greeting + +### Check status (use statusQueryGetUri from response) +GET {{baseUrl}}/instances/{{greeting.response.body.id}} + + +# --- TEST 2: Fan-out/fan-in --- +# Expected: Completed with ["Hello, Tokyo!", "Hello, London!", "Hello, Seattle!"] + +### Start fan-out orchestration +# @name fanout +POST {{baseUrl}}/orchestrators/fanout + +### Check status +GET {{baseUrl}}/instances/{{fanout.response.body.id}} + + +# --- TEST 3: Sub-orchestration --- +# Expected: Completed with "Parent received: Hello, World!" + +### Start parent orchestration +# @name parent +POST {{baseUrl}}/orchestrators/parent + +### Check status +GET {{baseUrl}}/instances/{{parent.response.body.id}} + + +# --- TEST 4: Entity interaction --- +# Expected: Completed with 30 + +### Start counter orchestration +# @name counter +POST {{baseUrl}}/orchestrators/counter + +### Check status +GET {{baseUrl}}/instances/{{counter.response.body.id}} + + +# --- TEST 5: Filter isolation --- +# Expected: Stays Pending (no worker has this function) + +### Start unknown orchestration +# @name unknown +POST {{baseUrl}}/start/SomeOtherOrchestration + +### Check — should be Pending +GET {{baseUrl}}/instances/{{unknown.response.body.id}} + + +# --- EXPECTED RESULTS --- +# +# | Test | Orchestration | Status | Why | +# |------|--------------------------|-----------|---------------------------| +# | 1 | greeting_orchestration | Completed | Matches filter | +# | 2 | fan_out_orchestration | Completed | Activities also filtered | +# | 3 | parent_orchestration | Completed | Sub-orch also filtered | +# | 4 | counter_orchestration | Completed | Entities also filtered | +# | 5 | SomeOtherOrchestration | Pending | No matching filter → held |