supported-patterns.md

Supported patterns

The following orchestration patterns are currently supported.

Function chaining

An orchestration can chain a sequence of function calls using the following syntax:

# simple activity function that returns a greetingdefhello(ctx: task.ActivityContext, name: str) ->str:
returnf'Hello {name}!'# orchestrator function that sequences the activity callsdefsequence(ctx: task.OrchestrationContext, _):
result1=yieldctx.call_activity(hello, input='Tokyo')
result2=yieldctx.call_activity(hello, input='Seattle')
result3=yieldctx.call_activity(hello, input='London')
return [result1, result2, result3]

See the full function chaining example.

Fan-out/fan-in

An orchestration can fan-out a dynamic number of function calls in parallel and then fan-in the results using the following syntax:

# activity function for getting the list of work itemsdefget_work_items(ctx: task.ActivityContext, _) ->List[str]:
# ...# activity function for processing a single work itemdefprocess_work_item(ctx: task.ActivityContext, item: str) ->int:
# ...# orchestrator function that fans-out the work items and then fans-in the resultsdeforchestrator(ctx: task.OrchestrationContext, _):
# the number of work-items is unknown in advancework_items=yieldctx.call_activity(get_work_items)
# fan-out: schedule the work items in parallel and wait for all of them to completetasks= [ctx.call_activity(process_work_item, input=item) foriteminwork_items]
results=yieldtask.when_all(tasks)
# fan-in: summarize and return the resultsreturn {'work_items': work_items, 'results': results, 'total': sum(results)}

See the full fan-out sample.

Human interaction and durable timers

An orchestration can wait for a user-defined event, such as a human approval event, before proceeding to the next step. In addition, the orchestration can create a timer with an arbitrary duration that triggers some alternate action if the external event hasn't been received:

defpurchase_order_workflow(ctx: task.OrchestrationContext, order: Order):
"""Orchestrator function that represents a purchase order workflow"""# Orders under $1000 are auto-approvediforder.Cost<1000:
return"Auto-approved"# Orders of $1000 or more require manager approvalyieldctx.call_activity(send_approval_request, input=order)
# Approvals must be received within 24 hours or they will be cancelled.# Passing ``data_type`` reconstructs the event payload as an ``Approval``.approval_event=ctx.wait_for_external_event("approval_received", data_type=Approval)
timeout_event=ctx.create_timer(timedelta(hours=24))
winner=yieldtask.when_any([approval_event, timeout_event])
ifwinner==timeout_event:
return"Cancelled"# The order was approvedyieldctx.call_activity(place_order, input=order)
approval_details=approval_event.get_result()
returnf"Approved by '{approval_details.approver}'"

As an aside, you'll also notice that the example orchestration above works with custom business objects. Custom classes, data classes, and named tuples are serialized to plain JSON automatically. To reconstruct the original type on the receiving side, supply the type — for example via the data_type argument to wait_for_external_event (shown above), the return_type argument to call_activity / call_sub_orchestrator / call_entity, or by annotating the consuming function's input parameter. Without a type, the payload is returned as plain JSON (a dict or list).

See the full human interaction sample.

Version-aware orchestrator

When utilizing orchestration versioning, it is possible for an orchestrator to remain backwards-compatible with orchestrations created using the previously defined version. For instance, consider an orchestration defined with the following signature:

defmy_orchestrator(ctx: task.OrchestrationContext, order: Order):
"""Dummy orchestrator function illustrating old logic"""yieldctx.call_activity(activity_one)
yieldctx.call_activity(activity_two)
return"Success"

Assume that any orchestrations created using this orchestrator were versioned 1.0.0. If the signature of this method needs to be updated to call activity_three between the calls to activity_one and activity_two, ordinarily this would break any running orchestrations at the time of deployment. However, the following orchestrator will be able to process both orchestrations versioned 1.0.0 and 2.0.0 after the change:

defmy_orchestrator(ctx: task.OrchestrationContext, order: Order):
"""Version-aware dummy orchestrator capable of processing both old and new orchestrations"""yieldctx.call_activity(activity_one)
ifctx.version>'1.0.0':
yieldctx.call_activity(activity_three)
yieldctx.call_activity(activity_two)

Alternatively, if the orchestrator changes completely, the following syntax might be preferred:

defmy_orchestrator(ctx: task.OrchestrationContext, order: Order):
ifctx.version=='1.0.0':
yieldctx.call_activity(activity_one)
yieldctx.call_activity(activity_two)
return "Successyieldctx.call_activity(activity_one)
yieldctx.call_activity(activity_three)
yieldctx.call_activity(activity_two)
return"Success"

See the full version-aware orchestrator sample

Work item filtering

When running multiple workers against the same task hub, each worker can declare which work items it handles. The backend then dispatches only the matching orchestrations, activities, and entities, avoiding unnecessary round-trips. Filtering is opt-in and supports both auto-generated and explicit filter sets.

The simplest approach auto-generates filters from the worker's registry:

withDurableTaskSchedulerWorker(...) asw:
w.add_orchestrator(greeting_orchestrator)
w.add_activity(greet)
w.use_work_item_filters() # auto-generate from registryw.start()

For more control you can provide explicit filters, including version constraints:

fromdurabletask.workerimport (
WorkItemFilters,
OrchestrationWorkItemFilter,
ActivityWorkItemFilter,
)
w.use_work_item_filters(WorkItemFilters(
orchestrations=[
OrchestrationWorkItemFilter(
name="greeting_orchestrator",
versions=["2.0.0"],
),
],
activities=[
ActivityWorkItemFilter(name="greet"),
],
))

See the full work item filtering sample.

Large payload externalization

When orchestrations work with very large inputs, outputs, or event data, the payloads can exceed gRPC message size limits. The large payload externalization pattern transparently offloads these payloads to Azure Blob Storage and replaces them with compact reference tokens in the gRPC messages.

No changes are required in orchestrator or activity code. Simply install the optional dependency and configure a payload store on the worker and client:

fromdurabletask.extensions.azure_blob_payloadsimportBlobPayloadStore, BlobPayloadStoreOptionsfromdurabletask.azuremanaged.clientimportDurableTaskSchedulerClientfromdurabletask.azuremanaged.workerimportDurableTaskSchedulerWorker# Configure the blob payload storestore=BlobPayloadStore(BlobPayloadStoreOptions(
connection_string="DefaultEndpointsProtocol=https;...",
))
# Pass the store to both worker and clientwithDurableTaskSchedulerWorker(
host_address=endpoint, secure_channel=secure_channel,
taskhub=taskhub_name, token_credential=credential,
payload_store=store,
) asw:
w.add_orchestrator(my_orchestrator)
w.add_activity(process_large_data)
w.start()
c=DurableTaskSchedulerClient(
host_address=endpoint, secure_channel=secure_channel,
taskhub=taskhub_name, token_credential=credential,
payload_store=store,
)
# This large input is automatically externalized to blob storagelarge_input="x"*1_000_000# 1 MB stringinstance_id=c.schedule_new_orchestration(my_orchestrator, input=large_input)
state=c.wait_for_orchestration_completion(instance_id, timeout=60)

In this example, any payload exceeding the threshold (default 256 KiB) is compressed and uploaded to the configured Azure Blob container. When the worker or client reads the message, it downloads and decompresses the payload automatically.

See the full large payload example and feature documentation for configuration options and details.