1# Durable Task Scheduler — Python
2 
3## Learn More
4 
5- [Durable Task Scheduler documentation](https://learn.microsoft.com/azure/durable-task-scheduler/)
6- [Durable Functions Python guide](https://learn.microsoft.com/azure/azure-functions/durable/durable-functions-overview?tabs=python)
7 
8## Durable Functions Setup
9 
10### Required Packages
11 
12```txt
13# requirements.txt
14azure-functions
15azure-functions-durable
16azure-identity
17```
18 
19> **💡 Finding latest versions**: Run `pip index versions azure-functions-durable` or check [pypi.org/project/azure-functions-durable](https://pypi.org/project/azure-functions-durable/) for the latest stable release.
20 
21### host.json
22 
23```json
24{
25  "version": "2.0",
26  "extensions": {
27    "durableTask": {
28      "hubName": "default",
29      "storageProvider": {
30        "type": "durabletask-scheduler",
31        "connectionStringName": "DURABLE_TASK_SCHEDULER_CONNECTION_STRING"
32      }
33    }
34  },
35  "extensionBundle": {
36    "id": "Microsoft.Azure.Functions.ExtensionBundle",
37    "version": "[4.*, 5.0.0)"
38  }
39}
40```
41 
42> **💡 NOTE**: Python uses extension bundles, so the storage provider type is `durabletask-scheduler`. .NET isolated uses the NuGet package directly and requires `azureManaged` instead — see [dotnet.md](dotnet.md).
43 
44### local.settings.json
45 
46```json
47{
48  "IsEncrypted": false,
49  "Values": {
50    "FUNCTIONS_WORKER_RUNTIME": "python",
51    "AzureWebJobsStorage": "UseDevelopmentStorage=true",
52    "DURABLE_TASK_SCHEDULER_CONNECTION_STRING": "Endpoint=http://localhost:8080;TaskHub=default;Authentication=None"
53  }
54}
55```
56 
57## Minimal Example
58 
59```python
60import azure.functions as func
61import azure.durable_functions as df
62 
63my_app = df.DFApp(http_auth_level=func.AuthLevel.FUNCTION)
64 
65# HTTP Starter
66@my_app.route(route="orchestrators/{function_name}", methods=["POST"])
67@my_app.durable_client_input(client_name="client")
68async def http_start(req: func.HttpRequest, client):
69    function_name = req.route_params.get('function_name')
70    instance_id = await client.start_new(function_name)
71    return client.create_check_status_response(req, instance_id)
72 
73# Orchestrator
74@my_app.orchestration_trigger(context_name="context")
75def my_orchestration(context: df.DurableOrchestrationContext):
76    result1 = yield context.call_activity("say_hello", "Tokyo")
77    result2 = yield context.call_activity("say_hello", "Seattle")
78    return f"{result1}, {result2}"
79 
80# Activity
81@my_app.activity_trigger(input_name="name")
82def say_hello(name: str) -> str:
83    return f"Hello {name}!"
84```
85 
86## Workflow Patterns
87 
88### Fan-Out/Fan-In
89 
90```python
91@my_app.orchestration_trigger(context_name="context")
92def fan_out_fan_in(context: df.DurableOrchestrationContext):
93    cities = ["Tokyo", "Seattle", "London", "Paris", "Berlin"]
94    
95    # Fan-out: schedule all in parallel
96    parallel_tasks = []
97    for city in cities:
98        task = context.call_activity("say_hello", city)
99        parallel_tasks.append(task)
100    
101    # Fan-in: wait for all
102    results = yield context.task_all(parallel_tasks)
103    return results
104```
105 
106### Human Interaction
107 
108```python
109import datetime
110 
111@my_app.orchestration_trigger(context_name="context")
112def approval_workflow(context: df.DurableOrchestrationContext):
113    yield context.call_activity("send_approval_request", context.get_input())
114    
115    # Wait for approval event with timeout
116    timeout = context.current_utc_datetime + datetime.timedelta(days=3)
117    approval_task = context.wait_for_external_event("ApprovalEvent")
118    timeout_task = context.create_timer(timeout)
119    
120    winner = yield context.task_any([approval_task, timeout_task])
121    
122    if winner == approval_task:
123        approved = approval_task.result
124        return "Approved" if approved else "Rejected"
125    return "Timed out"
126```
127 
128## Orchestration Determinism
129 
130| ❌ NEVER | ✅ ALWAYS USE |
131|----------|--------------|
132| `datetime.now()` | `context.current_utc_datetime` |
133| `uuid.uuid4()` | `context.new_uuid()` |
134| `random.random()` | Pass random values from activities |
135| `time.sleep()` | `context.create_timer()` |
136| Direct I/O, HTTP, database | `context.call_activity()` |
137 
138### Replay-Safe Logging
139 
140```python
141import logging
142 
143@my_app.orchestration_trigger(context_name="context")
144def my_orchestration(context: df.DurableOrchestrationContext):
145    # Check if replaying to avoid duplicate logs
146    if not context.is_replaying:
147        logging.info("Started")  # Only logs once, not on replay
148    result = yield context.call_activity("my_activity", "input")
149    return result
150```
151 
152## Error Handling & Retry
153 
154```python
155retry_options = df.RetryOptions(
156    first_retry_interval_in_milliseconds=5000,
157    max_number_of_attempts=3,
158    backoff_coefficient=2.0,
159    max_retry_interval_in_milliseconds=60000
160)
161 
162@my_app.orchestration_trigger(context_name="context")
163def workflow_with_retry(context: df.DurableOrchestrationContext):
164    try:
165        result = yield context.call_activity_with_retry(
166            "unreliable_service", 
167            retry_options, 
168            context.get_input()
169        )
170        return result
171    except Exception as ex:
172        context.set_custom_status({"error": str(ex)})
173        yield context.call_activity("compensation_activity", context.get_input())
174        return "Compensated"
175```
176 
177## Durable Task SDK (Non-Functions)
178 
179For applications running outside Azure Functions (containers, VMs, Azure Container Apps, Azure Kubernetes Service):
180 
181```python
182import asyncio
183from durabletask.azuremanaged.worker import DurableTaskSchedulerWorker
184 
185# Activity function
186def say_hello(ctx, name: str) -> str:
187    return f"Hello {name}!"
188 
189# Orchestrator function
190def my_orchestration(ctx, name: str) -> str:
191    result = yield ctx.call_activity('say_hello', input=name)
192    return result
193 
194async def main():
195    with DurableTaskSchedulerWorker(
196        host_address="http://localhost:8080",
197        secure_channel=False,
198        taskhub="default"
199    ) as worker:
200        worker.add_activity(say_hello)
201        worker.add_orchestrator(my_orchestration)
202        worker.start()
203 
204        # Client
205        from durabletask.azuremanaged.client import DurableTaskSchedulerClient
206        client = DurableTaskSchedulerClient(
207            host_address="http://localhost:8080",
208            taskhub="default",
209            token_credential=None,
210            secure_channel=False
211        )
212        instance_id = client.schedule_new_orchestration("my_orchestration", input="World")
213        result = client.wait_for_orchestration_completion(instance_id, timeout=30)
214        print(f"Output: {result.serialized_output}")
215 
216if __name__ == "__main__":
217    asyncio.run(main())
218```
219 
220