diff --git a/.github/workflows/build-samples.yml b/.github/workflows/build-samples.yml index a68f0cc..f670328 100644 --- a/.github/workflows/build-samples.yml +++ b/.github/workflows/build-samples.yml @@ -105,6 +105,12 @@ jobs: - name: Build DF - LargePayloadFanOutFanIn run: dotnet build samples/durable-functions/dotnet/LargePayloadFanOutFanIn/LargePayloadFanOutFanIn.csproj + - name: Build DF - WorkItemFiltering + run: dotnet build samples/durable-functions/dotnet/WorkItemFiltering/WorkItemFiltering.csproj + + - name: Build DF - WorkItemFiltering.AppB + run: dotnet build samples/durable-functions/dotnet/WorkItemFiltering.AppB/AppB.csproj + # Durable Functions Aspire samples (net10.0) - name: Build DF - AzureFunctionsAndDtsWithAspire run: dotnet build samples/durable-functions/dotnet/AzureFunctionsAndDtsWithAspire/AspireHost/AspireHost.csproj @@ -209,6 +215,12 @@ jobs: - name: Install DF - pdf-summarizer run: pip install -r samples/durable-functions/python/pdf-summarizer/requirements.txt + - name: Install DF - work-item-filtering + run: pip install -r samples/durable-functions/python/work-item-filtering/requirements.txt + + - name: Install DF - work-item-filtering-app-b + run: pip install -r samples/durable-functions/python/work-item-filtering-app-b/requirements.txt + # Syntax check all Python files - name: Syntax check Python files run: find samples -name "*.py" -not -path "*/__pycache__/*" -exec python -m py_compile {} + diff --git a/samples/durable-functions/dotnet/WorkItemFiltering.AppB/AppB.csproj b/samples/durable-functions/dotnet/WorkItemFiltering.AppB/AppB.csproj new file mode 100644 index 0000000..9d45ea3 --- /dev/null +++ b/samples/durable-functions/dotnet/WorkItemFiltering.AppB/AppB.csproj @@ -0,0 +1,18 @@ + + + net8.0 + v4 + Exe + enable + enable + WorkItemFiltering.AppB + WorkItemFiltering.AppB + + + + + + + + + diff --git a/samples/durable-functions/dotnet/WorkItemFiltering.AppB/Functions.cs b/samples/durable-functions/dotnet/WorkItemFiltering.AppB/Functions.cs new file mode 100644 index 0000000..ea5e534 --- /dev/null +++ b/samples/durable-functions/dotnet/WorkItemFiltering.AppB/Functions.cs @@ -0,0 +1,72 @@ +// 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.Extensions.Logging; + +namespace WorkItemFiltering.AppB; + +// ============================================================================= +// App B — registers an entirely DIFFERENT set of functions from App A. +// Both apps share the same DTS task hub ("default"). Work item filtering ensures +// each app only receives work items for the functions it has registered. +// +// App A owns: GreetingOrchestration, FanOutOrchestration, ParentOrchestration, +// CounterOrchestration, SayHello activity, CounterEntity +// App B owns: OrdersOrchestration, ShipOrder activity +// +// Either app's client endpoint can SCHEDULE any orchestration name. The +// scheduler routes the work item to the app whose filter matches. +// ============================================================================= + +public static class OrdersOrchestration +{ + [Function(nameof(OrdersOrchestration))] + public static async Task Run([OrchestrationTrigger] TaskOrchestrationContext ctx) + { + var logger = ctx.CreateReplaySafeLogger(nameof(OrdersOrchestration)); + logger.LogInformation("OrdersOrchestration started on App B"); + + string orderId = ctx.GetInput() ?? $"order-{ctx.NewGuid():N}"; + string shipResult = await ctx.CallActivityAsync(nameof(ShipOrder), orderId); + return shipResult; + } + + [Function(nameof(OrdersOrchestration) + "_Start")] + public static async Task Start( + [HttpTrigger(AuthorizationLevel.Anonymous, "post", Route = "orchestrators/orders")] HttpRequestData req, + [DurableClient] DurableTaskClient client) + { + string instanceId = await client.ScheduleNewOrchestrationInstanceAsync( + nameof(OrdersOrchestration), input: "order-42"); + return client.CreateCheckStatusResponse(req, instanceId); + } +} + +public static class ShipOrder +{ + [Function(nameof(ShipOrder))] + public static string Run([ActivityTrigger] string orderId, FunctionContext ctx) + { + ctx.GetLogger(nameof(ShipOrder)).LogInformation("App B shipping {OrderId}", orderId); + return $"Shipped {orderId} from App B"; + } +} + +// Generic starter so you can schedule ANY orchestration name from App B's port too. +public static class GenericStarter +{ + [Function("AppB_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.AppB/Program.cs b/samples/durable-functions/dotnet/WorkItemFiltering.AppB/Program.cs new file mode 100644 index 0000000..46bca8a --- /dev/null +++ b/samples/durable-functions/dotnet/WorkItemFiltering.AppB/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.AppB/host.json b/samples/durable-functions/dotnet/WorkItemFiltering.AppB/host.json new file mode 100644 index 0000000..cf8d8c4 --- /dev/null +++ b/samples/durable-functions/dotnet/WorkItemFiltering.AppB/host.json @@ -0,0 +1,22 @@ +{ + "version": "2.0", + "logging": { + "logLevel": { + "default": "Information", + "DurableTask.AzureStorage": "Warning", + "DurableTask.Core": "Warning", + "Microsoft.DurableTask": "Information", + "Host.Triggers.DurableTask": "Information" + } + }, + "extensions": { + "durableTask": { + "hubName": "default", + "storageProvider": { + "type": "azureManaged", + "connectionStringName": "DURABLE_TASK_SCHEDULER_CONNECTION_STRING", + "workItemFilteringEnabled": true + } + } + } +} diff --git a/samples/durable-functions/dotnet/WorkItemFiltering/Functions.cs b/samples/durable-functions/dotnet/WorkItemFiltering/Functions.cs new file mode 100644 index 0000000..bca4943 --- /dev/null +++ b/samples/durable-functions/dotnet/WorkItemFiltering/Functions.cs @@ -0,0 +1,161 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. + +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..1563e10 --- /dev/null +++ b/samples/durable-functions/dotnet/WorkItemFiltering/README.md @@ -0,0 +1,151 @@ +# 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 + +A sibling project [`WorkItemFiltering.AppB`](../WorkItemFiltering.AppB/) registers an entirely **different** set of functions (`OrdersOrchestration`, `ShipOrder` activity) against the **same** DTS task hub (`default`). The scheduler routes each work item to whichever app's filter matches. + +Run both apps together with the included script: + +```powershell +# From samples/durable-functions/dotnet/ +.\run-both.ps1 +``` + +The script builds both projects, starts App A on `:7071` and App B on `:7072`, and exercises five scenarios: + +| # | Scenario | Expected | +|---|----------------------------------------------------------------|---------------------| +| 1 | App A orchestration, scheduled from App A client | `Completed` | +| 2 | App B orchestration, scheduled from App B client | `Completed` | +| 3 | **Cross-app:** App B's `OrdersOrchestration` from App A client | `Completed` on B | +| 4 | **Cross-app:** App A's `GreetingOrchestration` from App B client | `Completed` on A | +| 5 | Orchestration neither app has registered | `Pending` (forever) | + +Scenarios 3 and 4 are the point: the client app does not need to host the orchestrator. Without `workItemFilteringEnabled`, both apps would race to dispatch every work item and one would fail with *"function does not exist"*. With filtering on, the scheduler delivers only matching work to each app. + +Helpful flags: + +```powershell +.\run-both.ps1 -StartOnly # leave both running for manual testing +.\run-both.ps1 -StopOnly # kill any running func hosts +``` + +## 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..2b80209 --- /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..cf8d8c4 --- /dev/null +++ b/samples/durable-functions/dotnet/WorkItemFiltering/host.json @@ -0,0 +1,22 @@ +{ + "version": "2.0", + "logging": { + "logLevel": { + "default": "Information", + "DurableTask.AzureStorage": "Warning", + "DurableTask.Core": "Warning", + "Microsoft.DurableTask": "Information", + "Host.Triggers.DurableTask": "Information" + } + }, + "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..d9912e2 --- /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 (uses statusQueryGetUri returned by the starter) +GET {{greeting.response.body.statusQueryGetUri}} + + +# --- 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 {{fanout.response.body.statusQueryGetUri}} + + +# --- TEST 3: Sub-orchestration --- +# Expected: Completed with "Parent received: Hello, World!" + +### Start ParentOrchestration +# @name parent +POST {{baseUrl}}/orchestrators/parent + +### Check status +GET {{parent.response.body.statusQueryGetUri}} + + +# --- TEST 4: Entity interaction --- +# Expected: Completed with 30 (10 + 20) + +### Start CounterOrchestration +# @name counter +POST {{baseUrl}}/orchestrators/counter + +### Check status +GET {{counter.response.body.statusQueryGetUri}} + + +# --- 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 {{unknown.response.body.statusQueryGetUri}} + +### Check again after 15 seconds — still Pending +GET {{unknown.response.body.statusQueryGetUri}} + + +# --- 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/dotnet/run-both.ps1 b/samples/durable-functions/dotnet/run-both.ps1 new file mode 100644 index 0000000..4f72619 --- /dev/null +++ b/samples/durable-functions/dotnet/run-both.ps1 @@ -0,0 +1,91 @@ +# Runs App A (port 7071) and App B (port 7072) against the same DTS task hub. +# Demonstrates work item filtering: each app only processes the orchestrations +# and activities it has registered. Cross-app scheduling still works because +# DTS routes each work item to the app whose filter matches. +# +# Usage: +# .\run-both.ps1 # start both apps, wait, exercise scenarios +# .\run-both.ps1 -StartOnly # just start them and leave running +# .\run-both.ps1 -StopOnly # kill any running func hosts on 7071/7072 + +[CmdletBinding()] +param( + [switch]$StartOnly, + [switch]$StopOnly +) + +$ErrorActionPreference = 'Stop' +$root = Split-Path -Parent $MyInvocation.MyCommand.Path +$appA = Join-Path $root 'WorkItemFiltering' +$appB = Join-Path $root 'WorkItemFiltering.AppB' +$logA = Join-Path $env:TEMP 'wif-appA.log' +$logB = Join-Path $env:TEMP 'wif-appB.log' + +function Stop-Funcs { + Get-Process func, Microsoft.Azure.Functions.JobHost -ErrorAction SilentlyContinue | + Stop-Process -Force -ErrorAction SilentlyContinue +} + +Stop-Funcs +if ($StopOnly) { Write-Host 'Stopped func hosts.'; return } + +Write-Host '== Building App A and App B ==' +dotnet build $appA -c Debug --nologo | Select-Object -Last 3 +dotnet build $appB -c Debug --nologo | Select-Object -Last 3 + +Remove-Item $logA, $logB -ErrorAction SilentlyContinue + +Write-Host '== Starting App A on :7071 ==' +$pA = Start-Process -FilePath func -ArgumentList 'start','--port','7071' ` + -WorkingDirectory $appA -RedirectStandardOutput $logA ` + -RedirectStandardError "$logA.err" -NoNewWindow -PassThru + +Write-Host '== Starting App B on :7072 ==' +$pB = Start-Process -FilePath func -ArgumentList 'start','--port','7072' ` + -WorkingDirectory $appB -RedirectStandardOutput $logB ` + -RedirectStandardError "$logB.err" -NoNewWindow -PassThru + +Write-Host "App A PID=$($pA.Id) App B PID=$($pB.Id)" +Write-Host 'Waiting 30s for both hosts to start and register filters...' +Start-Sleep 30 + +if ($StartOnly) { + Write-Host 'Both apps running. Stop with: .\run-both.ps1 -StopOnly' + return +} + +function Invoke-Scenario([string]$Name, [string]$Url) { + Write-Host "`n-- $Name --" + try { + $r = Invoke-RestMethod -Method Post -Uri $Url -TimeoutSec 30 + Start-Sleep 8 + $s = Invoke-RestMethod -Uri $r.statusQueryGetUri + Write-Host (" status={0} output={1}" -f $s.runtimeStatus, ($s.output | ConvertTo-Json -Compress)) + } catch { + Write-Host " ERROR: $_" + } +} + +Invoke-Scenario 'App A orchestration via App A client (own filter)' ` + 'http://localhost:7071/api/orchestrators/greeting' + +Invoke-Scenario 'App B orchestration via App B client (own filter)' ` + 'http://localhost:7072/api/orchestrators/orders' + +Invoke-Scenario 'CROSS-APP: schedule App B orchestration from App A client' ` + 'http://localhost:7071/api/start/OrdersOrchestration' + +Invoke-Scenario 'CROSS-APP: schedule App A orchestration from App B client' ` + 'http://localhost:7072/api/start/GreetingOrchestration' + +Invoke-Scenario 'UNKNOWN: orchestration that no app has registered' ` + 'http://localhost:7071/api/start/NobodyOwnsThis' + +Write-Host "`n== Filter registration log lines ==" +Write-Host '-- App A --' +Select-String -Path $logA -Pattern 'Work item filtering|Registered \d+ orch' | Select-Object -Last 2 +Write-Host '-- App B --' +Select-String -Path $logB -Pattern 'Work item filtering|Registered \d+ orch' | Select-Object -Last 2 + +Write-Host "`nLogs: $logA / $logB" +Write-Host 'Apps still running. Stop with: .\run-both.ps1 -StopOnly' diff --git a/samples/durable-functions/python/run-both.ps1 b/samples/durable-functions/python/run-both.ps1 new file mode 100644 index 0000000..2ace3fb --- /dev/null +++ b/samples/durable-functions/python/run-both.ps1 @@ -0,0 +1,119 @@ +# Runs work-item-filtering App A (port 7071) and App B (port 7072) against the +# same DTS task hub. Both apps have manual extensions (extensions.csproj +# referencing Microsoft.Azure.WebJobs.Extensions.DurableTask.AzureManaged +# v1.8.1 from the local NuGet feed) so they pick up the latest filter changes +# instead of the extension bundle. +# +# Usage: +# .\run-both.ps1 # set up venv, install extensions, run scenarios +# .\run-both.ps1 -StartOnly # set up & start, leave running +# .\run-both.ps1 -StopOnly # kill any running func hosts + +[CmdletBinding()] +param( + [switch]$StartOnly, + [switch]$StopOnly, + [switch]$SkipSetup +) + +$ErrorActionPreference = 'Stop' +$root = Split-Path -Parent $MyInvocation.MyCommand.Path +$appA = Join-Path $root 'work-item-filtering' +$appB = Join-Path $root 'work-item-filtering-app-b' +$venv = Join-Path $root '.venv-wif' +$logA = Join-Path $env:TEMP 'wif-py-appA.log' +$logB = Join-Path $env:TEMP 'wif-py-appB.log' + +function Stop-Funcs { + Get-Process func, Microsoft.Azure.Functions.JobHost, python -ErrorAction SilentlyContinue | + Where-Object { + $_.ProcessName -in @('func','Microsoft.Azure.Functions.JobHost') -or + ($_.ProcessName -eq 'python' -and $_.Path -like "$venv*") + } | + Stop-Process -Force -ErrorAction SilentlyContinue +} + +Stop-Funcs +if ($StopOnly) { Write-Host 'Stopped func hosts.'; return } + +if (-not $SkipSetup) { + if (-not (Test-Path $venv)) { + Write-Host '== Creating shared venv ==' + python -m venv $venv + } + $py = Join-Path $venv 'Scripts\python.exe' + Write-Host '== Installing Python deps ==' + & $py -m pip install --quiet --upgrade pip + & $py -m pip install --quiet -r (Join-Path $appA 'requirements.txt') + + foreach ($app in @($appA, $appB)) { + Write-Host "== func extensions install in $(Split-Path -Leaf $app) ==" + Push-Location $app + try { + # func extensions install shells out to `dotnet build` of extensions.csproj + # The local NuGet.config feed will be picked up automatically. + func extensions install --force 2>&1 | Select-Object -Last 4 + } finally { Pop-Location } + } +} + +# Activate venv for this shell so Start-Process inherits the right python on PATH +$env:VIRTUAL_ENV = $venv +$env:PATH = "$venv\Scripts;$env:PATH" + +Remove-Item $logA, $logB, "$logA.err", "$logB.err" -ErrorAction SilentlyContinue + +Write-Host '== Starting App A on :7071 ==' +$pA = Start-Process -FilePath func -ArgumentList 'start','--port','7071' ` + -WorkingDirectory $appA -RedirectStandardOutput $logA ` + -RedirectStandardError "$logA.err" -NoNewWindow -PassThru + +Write-Host '== Starting App B on :7072 ==' +$pB = Start-Process -FilePath func -ArgumentList 'start','--port','7072' ` + -WorkingDirectory $appB -RedirectStandardOutput $logB ` + -RedirectStandardError "$logB.err" -NoNewWindow -PassThru + +Write-Host "App A PID=$($pA.Id) App B PID=$($pB.Id)" +Write-Host 'Waiting 30s for both hosts to start and register filters...' +Start-Sleep 30 + +if ($StartOnly) { + Write-Host 'Both apps running. Stop with: .\run-both.ps1 -StopOnly' + return +} + +function Invoke-Scenario([string]$Name, [string]$Url) { + Write-Host "`n-- $Name --" + try { + $r = Invoke-RestMethod -Method Post -Uri $Url -TimeoutSec 30 + Start-Sleep 8 + $s = Invoke-RestMethod -Uri $r.statusQueryGetUri + Write-Host (" status={0} output={1}" -f $s.runtimeStatus, ($s.output | ConvertTo-Json -Compress)) + } catch { + Write-Host " ERROR: $_" + } +} + +Invoke-Scenario 'App A orchestration via App A client (own filter)' ` + 'http://localhost:7071/api/orchestrators/greeting' + +Invoke-Scenario 'App B orchestration via App B client (own filter)' ` + 'http://localhost:7072/api/orchestrators/orders' + +Invoke-Scenario 'CROSS-APP: schedule App B orchestration from App A client' ` + 'http://localhost:7071/api/start/orders_orchestration' + +Invoke-Scenario 'CROSS-APP: schedule App A orchestration from App B client' ` + 'http://localhost:7072/api/start/greeting_orchestration' + +Invoke-Scenario 'UNKNOWN: orchestration that no app has registered' ` + 'http://localhost:7071/api/start/nobody_owns_this' + +Write-Host "`n== Filter registration log lines ==" +Write-Host '-- App A --' +Select-String -Path $logA -Pattern 'Work item filtering|Registered \d+ orch|AzureManagedProvider' | Select-Object -Last 3 +Write-Host '-- App B --' +Select-String -Path $logB -Pattern 'Work item filtering|Registered \d+ orch|AzureManagedProvider' | Select-Object -Last 3 + +Write-Host "`nLogs: $logA / $logB" +Write-Host 'Apps still running. Stop with: .\run-both.ps1 -StopOnly' diff --git a/samples/durable-functions/python/work-item-filtering-app-b/extensions.csproj b/samples/durable-functions/python/work-item-filtering-app-b/extensions.csproj new file mode 100644 index 0000000..0c35d3f --- /dev/null +++ b/samples/durable-functions/python/work-item-filtering-app-b/extensions.csproj @@ -0,0 +1,13 @@ + + + net8.0 + + ** + <_FunctionsSkipCleanOutput>true + + + + + + + diff --git a/samples/durable-functions/python/work-item-filtering-app-b/function_app.py b/samples/durable-functions/python/work-item-filtering-app-b/function_app.py new file mode 100644 index 0000000..0d0703c --- /dev/null +++ b/samples/durable-functions/python/work-item-filtering-app-b/function_app.py @@ -0,0 +1,57 @@ +# Copyright (c) Microsoft Corporation. All rights reserved. +# Licensed under the MIT License. + +""" +Work Item Filtering — App B (Python). + +App B registers an entirely DIFFERENT set of functions from App A. Both apps +share the same DTS task hub ("default"). Work item filtering ensures each app +only receives work items for the functions IT has registered. + +App A owns: greeting_orchestration, fan_out_orchestration, + parent_orchestration, counter_orchestration, + say_hello activity, counter_entity +App B owns: orders_orchestration, ship_order activity + +Either app's HTTP client can SCHEDULE any orchestration name. The scheduler +routes the work item to the app whose filter matches. +""" + +import logging +import azure.functions as func +import azure.durable_functions as df + +app = func.FunctionApp(http_auth_level=func.AuthLevel.ANONYMOUS) +bp = df.Blueprint() + + +@bp.orchestration_trigger(context_name="context") +def orders_orchestration(context: df.DurableOrchestrationContext): + order_id = context.get_input() or f"order-{context.new_guid()}" + result = yield context.call_activity("ship_order", order_id) + return result + + +@bp.activity_trigger(input_name="order_id") +def ship_order(order_id: str) -> str: + logging.info("App B shipping %s", order_id) + return f"Shipped {order_id} from App B" + + +@app.route(route="orchestrators/orders", methods=["POST"]) +@app.durable_client_input(client_name="client") +async def start_orders(req: func.HttpRequest, client) -> func.HttpResponse: + instance_id = await client.start_new("orders_orchestration", client_input="order-42") + return client.create_check_status_response(req, instance_id) + + +@app.route(route="start/{name}", methods=["POST"]) +@app.durable_client_input(client_name="client") +async def start_any(req: func.HttpRequest, client) -> func.HttpResponse: + """Generic starter: schedule ANY orchestration by name from App B.""" + name = req.route_params.get("name") + 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-app-b/host.json b/samples/durable-functions/python/work-item-filtering-app-b/host.json new file mode 100644 index 0000000..cf8d8c4 --- /dev/null +++ b/samples/durable-functions/python/work-item-filtering-app-b/host.json @@ -0,0 +1,22 @@ +{ + "version": "2.0", + "logging": { + "logLevel": { + "default": "Information", + "DurableTask.AzureStorage": "Warning", + "DurableTask.Core": "Warning", + "Microsoft.DurableTask": "Information", + "Host.Triggers.DurableTask": "Information" + } + }, + "extensions": { + "durableTask": { + "hubName": "default", + "storageProvider": { + "type": "azureManaged", + "connectionStringName": "DURABLE_TASK_SCHEDULER_CONNECTION_STRING", + "workItemFilteringEnabled": true + } + } + } +} diff --git a/samples/durable-functions/python/work-item-filtering-app-b/requirements.txt b/samples/durable-functions/python/work-item-filtering-app-b/requirements.txt new file mode 100644 index 0000000..4d2b03a --- /dev/null +++ b/samples/durable-functions/python/work-item-filtering-app-b/requirements.txt @@ -0,0 +1,2 @@ +azure-functions +azure-functions-durable 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..7b63ee7 --- /dev/null +++ b/samples/durable-functions/python/work-item-filtering/README.md @@ -0,0 +1,172 @@ +# 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. [.NET 8 SDK](https://dotnet.microsoft.com/download/dotnet/8.0) — required to build the manual extension bundle (see below) +3. [Docker](https://www.docker.com/products/docker-desktop/) (for running the emulator and Azurite) +4. [Azure Functions Core Tools v4](https://learn.microsoft.com/azure/azure-functions/functions-run-local) + +## Extension Management + +This sample does **not** use the Azure Functions extension bundle. Instead it pins the Durable Task `azureManaged` provider to a specific version via [`extensions.csproj`](extensions.csproj) — required so the app picks up updates to `Microsoft.Azure.WebJobs.Extensions.DurableTask.AzureManaged` ahead of the public bundle. + +Before the first run, install the extensions into `bin/`: + +```powershell +func extensions install +``` + +Core Tools reads `extensions.csproj`, restores the pinned packages, and generates `bin/extensions.json` with the Durable + AzureManaged DLLs so the Functions host loads them at startup. + +## 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, build the extensions, 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 extensions install + 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 + +## Multi-App Scenario + +A sibling project [`work-item-filtering-app-b`](../work-item-filtering-app-b/) registers an entirely **different** set of functions (`orders_orchestration`, `ship_order` activity) against the **same** DTS task hub (`default`). The scheduler routes each work item to whichever app's filter matches. + +Run both apps together with the included script: + +```powershell +# From samples/durable-functions/python/ +.\run-both.ps1 +``` + +The script: + +1. Creates a shared venv at `python/.venv-wif` and installs `requirements.txt` +2. Builds `extensions.csproj` for both apps (App A and App B) +3. Starts App A on `:7071` and App B on `:7072` +4. Exercises five scenarios: + +| # | Scenario | Expected | +|---|----------------------------------------------------------------|---------------------| +| 1 | App A orchestration, scheduled from App A client | `Completed` | +| 2 | App B orchestration, scheduled from App B client | `Completed` | +| 3 | **Cross-app:** App B's `orders_orchestration` from App A client | `Completed` on B | +| 4 | **Cross-app:** App A's `greeting_orchestration` from App B client | `Completed` on A | +| 5 | Orchestration neither app has registered | `Pending` (forever) | + +Scenarios 3 and 4 are the point: the client app does not need to host the orchestrator. Without `workItemFilteringEnabled`, both apps would race to dispatch every work item and one would fail with *"function does not exist"*. With filtering on, the scheduler delivers only matching work to each app. + +Helpful flags: + +```powershell +.\run-both.ps1 -StartOnly # leave both running for manual testing +.\run-both.ps1 -StopOnly # kill any running func hosts +.\run-both.ps1 -SkipSetup # skip venv + extension build +``` + +## 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/extensions.csproj b/samples/durable-functions/python/work-item-filtering/extensions.csproj new file mode 100644 index 0000000..0c35d3f --- /dev/null +++ b/samples/durable-functions/python/work-item-filtering/extensions.csproj @@ -0,0 +1,13 @@ + + + net8.0 + + ** + <_FunctionsSkipCleanOutput>true + + + + + + + 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..8e97e48 --- /dev/null +++ b/samples/durable-functions/python/work-item-filtering/function_app.py @@ -0,0 +1,142 @@ +# 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 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 +# ============================================================================= + +@app.route(route="orchestrators/greeting", methods=["POST"]) +@app.durable_client_input(client_name="client") +async def start_greeting(req: func.HttpRequest, client) -> func.HttpResponse: + """Start the greeting orchestration.""" + instance_id = await client.start_new("greeting_orchestration") + return client.create_check_status_response(req, instance_id) + + +@app.route(route="orchestrators/fanout", methods=["POST"]) +@app.durable_client_input(client_name="client") +async def start_fan_out(req: func.HttpRequest, client) -> func.HttpResponse: + """Start the fan-out orchestration.""" + instance_id = await client.start_new("fan_out_orchestration") + return client.create_check_status_response(req, instance_id) + + +@app.route(route="orchestrators/parent", methods=["POST"]) +@app.durable_client_input(client_name="client") +async def start_parent(req: func.HttpRequest, client) -> func.HttpResponse: + """Start the parent orchestration.""" + instance_id = await client.start_new("parent_orchestration") + return client.create_check_status_response(req, instance_id) + + +@app.route(route="orchestrators/counter", methods=["POST"]) +@app.durable_client_input(client_name="client") +async def start_counter(req: func.HttpRequest, client) -> func.HttpResponse: + """Start the counter orchestration (entity interaction).""" + instance_id = await client.start_new("counter_orchestration") + return client.create_check_status_response(req, instance_id) + + +@app.route(route="start/{name}", methods=["POST"]) +@app.durable_client_input(client_name="client") +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") + 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..cf8d8c4 --- /dev/null +++ b/samples/durable-functions/python/work-item-filtering/host.json @@ -0,0 +1,22 @@ +{ + "version": "2.0", + "logging": { + "logLevel": { + "default": "Information", + "DurableTask.AzureStorage": "Warning", + "DurableTask.Core": "Warning", + "Microsoft.DurableTask": "Information", + "Host.Triggers.DurableTask": "Information" + } + }, + "extensions": { + "durableTask": { + "hubName": "default", + "storageProvider": { + "type": "azureManaged", + "connectionStringName": "DURABLE_TASK_SCHEDULER_CONNECTION_STRING", + "workItemFilteringEnabled": true + } + } + } +} 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..4e578b0 --- /dev/null +++ b/samples/durable-functions/python/work-item-filtering/test.http @@ -0,0 +1,82 @@ +# ============================================================================= +# 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: +# Linux/macOS: python -m venv .venv && source .venv/bin/activate && pip install -r requirements.txt && func extensions install && func start +# Windows: python -m venv .venv && .\.venv\Scripts\activate && pip install -r requirements.txt && func extensions install && 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 (uses statusQueryGetUri returned by the starter) +GET {{greeting.response.body.statusQueryGetUri}} + + +# --- 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 {{fanout.response.body.statusQueryGetUri}} + + +# --- TEST 3: Sub-orchestration --- +# Expected: Completed with "Parent received: Hello, World!" + +### Start parent orchestration +# @name parent +POST {{baseUrl}}/orchestrators/parent + +### Check status +GET {{parent.response.body.statusQueryGetUri}} + + +# --- TEST 4: Entity interaction --- +# Expected: Completed with 30 + +### Start counter orchestration +# @name counter +POST {{baseUrl}}/orchestrators/counter + +### Check status +GET {{counter.response.body.statusQueryGetUri}} + + +# --- 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 {{unknown.response.body.statusQueryGetUri}} + + +# --- 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 |