fixes, lint, remove grpc changes

Signed-off-by: Filinto Duran <[email protected]>

Author: filintod

dapr/conf/__init__.py

@@ -24,9 +24,7 @@ def __init__(self):
             default_value = getattr(global_settings, setting)
             env_variable = os.environ.get(setting)
             if env_variable:
-                val = (
-                    type(default_value)(env_variable) if default_value is not None else env_variable
-                )
+                val = self._coerce_env_value(default_value, env_variable)
                 setattr(self, setting, val)
             else:
                 setattr(self, setting, default_value)
@@ -36,5 +34,27 @@ def __getattr__(self, name):
             raise AttributeError(f"'{self.__class__.__name__}' object has no attribute '{name}'")
         return getattr(self, name)
 
+    @staticmethod
+    def _coerce_env_value(default_value, env_variable: str):
+        if default_value is None:
+            return env_variable
+        # Handle booleans explicitly to avoid bool('false') == True
+        if isinstance(default_value, bool):
+            s = env_variable.strip().lower()
+            if s in ('1', 'true', 't', 'yes', 'y', 'on'):
+                return True
+            if s in ('0', 'false', 'f', 'no', 'n', 'off'):
+                return False
+            # Fallback: non-empty -> True for backward-compat
+            return bool(s)
+        # Integers
+        if isinstance(default_value, int) and not isinstance(default_value, bool):
+            return int(env_variable)
+        # Floats
+        if isinstance(default_value, float):
+            return float(env_variable)
+        # Other types: try to cast as before
+        return type(default_value)(env_variable)
+
 
 settings = Settings()

dapr/conf/global_settings.py

@@ -34,6 +34,23 @@
 
 DAPR_HTTP_TIMEOUT_SECONDS = 60
 
+# gRPC keepalive (disabled by default; enable via env to help with idle debugging sessions)
+DAPR_GRPC_KEEPALIVE_ENABLED: bool = False
+DAPR_GRPC_KEEPALIVE_TIME_MS: int = 120000  # send keepalive pings every 120s
+DAPR_GRPC_KEEPALIVE_TIMEOUT_MS: int = (
+    20000  # wait 20s for ack before considering the connection dead
+)
+DAPR_GRPC_KEEPALIVE_PERMIT_WITHOUT_CALLS: bool = False  # allow pings when there are no active calls
+
+# gRPC retries (disabled by default; enable via env to apply channel service config)
+DAPR_GRPC_RETRY_ENABLED: bool = False
+DAPR_GRPC_RETRY_MAX_ATTEMPTS: int = 4
+DAPR_GRPC_RETRY_INITIAL_BACKOFF_MS: int = 100
+DAPR_GRPC_RETRY_MAX_BACKOFF_MS: int = 1000
+DAPR_GRPC_RETRY_BACKOFF_MULTIPLIER: float = 2.0
+# Comma-separated list of status codes, e.g., 'UNAVAILABLE,DEADLINE_EXCEEDED'
+DAPR_GRPC_RETRY_CODES: str = 'UNAVAILABLE,DEADLINE_EXCEEDED'
+
 # ----- Conversation API settings ------
 
 # Configuration for handling large enums to avoid massive JSON schemas that can exceed LLM token limits

dev-requirements.txt

@@ -2,7 +2,9 @@ mypy>=1.2.0
 mypy-extensions>=0.4.3
 mypy-protobuf>=2.9
 tox>=4.3.0
+pip>=23.0.0
 coverage>=5.3
+pytest
 wheel
 # used in unit test only
 opentelemetry-sdk

examples/workflow-async/README.md

@@ -0,0 +1,15 @@
+# Dapr Workflow Async Examples (Python)
+
+These examples mirror `examples/workflow/` but author orchestrators with `async def` using the
+async workflow APIs. Activities remain regular functions unless noted.
+
+How to run:
+- Ensure a Dapr sidecar is running locally. If needed, set `DURABLETASK_GRPC_ENDPOINT`, or
+  `DURABLETASK_GRPC_HOST/PORT`.
+- Install requirements: `pip install -r requirements.txt`
+- Run any example: `python simple.py`
+
+Notes:
+- Orchestrators use `await ctx.activity(...)`, `await ctx.sleep(...)`, `await ctx.when_all/when_any(...)`, etc.
+- No event loop is started manually; the Durable Task worker drives the async orchestrators.
+- You can also launch instances using `DaprWorkflowClient` as in the non-async examples.

examples/workflow-async/child_workflow.py

@@ -0,0 +1,47 @@
+# -*- coding: utf-8 -*-
+
+"""
+Copyright 2025 The Dapr Authors
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+    http://www.apache.org/licenses/LICENSE-2.0
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the specific language governing permissions and
+limitations under the License.
+"""
+
+from dapr.ext.workflow import (
+    AsyncWorkflowContext,
+    DaprWorkflowClient,
+    WorkflowRuntime,
+)
+
+wfr = WorkflowRuntime()
+
+
+@wfr.async_workflow(name='child_async')
+async def child(ctx: AsyncWorkflowContext, n: int) -> int:
+    return n * 2
+
+
+@wfr.async_workflow(name='parent_async')
+async def parent(ctx: AsyncWorkflowContext, n: int) -> int:
+    r = await ctx.call_child_workflow(child, input=n)
+    print(f'Child workflow returned {r}')
+    return r + 1
+
+
+def main():
+    wfr.start()
+    client = DaprWorkflowClient()
+    instance_id = 'parent_async_instance'
+    client.schedule_new_workflow(workflow=parent, input=5, instance_id=instance_id)
+    client.wait_for_workflow_completion(instance_id, timeout_in_seconds=60)
+    wfr.shutdown()
+
+
+if __name__ == '__main__':
+    main()

examples/workflow-async/fan_out_fan_in.py

@@ -0,0 +1,49 @@
+# -*- coding: utf-8 -*-
+"""
+Copyright 2025 The Dapr Authors
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+    http://www.apache.org/licenses/LICENSE-2.0
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the specific language governing permissions and
+limitations under the License.
+"""
+
+from dapr.ext.workflow import (
+    AsyncWorkflowContext,
+    DaprWorkflowClient,
+    WorkflowActivityContext,
+    WorkflowRuntime,
+)
+
+wfr = WorkflowRuntime()
+
+
+@wfr.activity(name='square')
+def square(ctx: WorkflowActivityContext, x: int) -> int:
+    return x * x
+
+
+@wfr.async_workflow(name='fan_out_fan_in_async')
+async def orchestrator(ctx: AsyncWorkflowContext):
+    tasks = [ctx.call_activity(square, input=i) for i in range(1, 6)]
+    results = await ctx.when_all(tasks)
+    total = sum(results)
+    return total
+
+
+def main():
+    wfr.start()
+    client = DaprWorkflowClient()
+    instance_id = 'fofi_async'
+    client.schedule_new_workflow(workflow=orchestrator, instance_id=instance_id)
+    wf_state = client.wait_for_workflow_completion(instance_id, timeout_in_seconds=60)
+    print(f'Workflow state: {wf_state}')
+    wfr.shutdown()
+
+
+if __name__ == '__main__':
+    main()

examples/workflow-async/human_approval.py

@@ -0,0 +1,47 @@
+# -*- coding: utf-8 -*-
+
+"""
+Copyright 2025 The Dapr Authors
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+    http://www.apache.org/licenses/LICENSE-2.0
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the specific language governing permissions and
+limitations under the License.
+"""
+
+from dapr.ext.workflow import AsyncWorkflowContext, DaprWorkflowClient, WorkflowRuntime
+
+wfr = WorkflowRuntime()
+
+
+@wfr.async_workflow(name='human_approval_async')
+async def orchestrator(ctx: AsyncWorkflowContext, request_id: str):
+    decision = await ctx.when_any(
+        [
+            ctx.wait_for_external_event(f'approve:{request_id}'),
+            ctx.wait_for_external_event(f'reject:{request_id}'),
+            ctx.create_timer(300.0),
+        ]
+    )
+    if isinstance(decision, dict) and decision.get('approved'):
+        return 'APPROVED'
+    if isinstance(decision, dict) and decision.get('rejected'):
+        return 'REJECTED'
+    return 'TIMEOUT'
+
+
+def main():
+    wfr.start()
+    client = DaprWorkflowClient()
+    instance_id = 'human_approval_async_1'
+    client.schedule_new_workflow(workflow=orchestrator, input='REQ-1', instance_id=instance_id)
+    # In a real scenario, raise approve/reject event from another service.
+    wfr.shutdown()
+
+
+if __name__ == '__main__':
+    main()

examples/workflow-async/requirements.txt

@@ -0,0 +1,2 @@
+dapr-ext-workflow-dev>=1.15.0.dev
+dapr-dev>=1.15.0.dev

examples/workflow-async/simple.py

@@ -0,0 +1,136 @@
+# -*- coding: utf-8 -*-
+"""
+Copyright 2025 The Dapr Authors
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+    http://www.apache.org/licenses/LICENSE-2.0
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the specific language governing permissions and
+limitations under the License.
+"""
+
+from datetime import timedelta
+from time import sleep
+
+from dapr.ext.workflow import (
+    AsyncWorkflowContext,
+    DaprWorkflowClient,
+    RetryPolicy,
+    WorkflowActivityContext,
+    WorkflowRuntime,
+)
+
+counter = 0
+retry_count = 0
+child_orchestrator_string = ''
+instance_id = 'asyncExampleInstanceID'
+child_instance_id = 'asyncChildInstanceID'
+workflow_name = 'async_hello_world_wf'
+child_workflow_name = 'async_child_wf'
+input_data = 'Hi Async Counter!'
+event_name = 'event1'
+event_data = 'eventData'
+
+retry_policy = RetryPolicy(
+    first_retry_interval=timedelta(seconds=1),
+    max_number_of_attempts=3,
+    backoff_coefficient=2,
+    max_retry_interval=timedelta(seconds=10),
+    retry_timeout=timedelta(seconds=100),
+)
+
+wfr = WorkflowRuntime()
+
+
+@wfr.async_workflow(name=workflow_name)
+async def hello_world_wf(ctx: AsyncWorkflowContext, wf_input):
+    # activities
+    result_1 = await ctx.call_activity(hello_act, input=1)
+    print(f'Activity 1 returned {result_1}')
+    result_2 = await ctx.call_activity(hello_act, input=10)
+    print(f'Activity 2 returned {result_2}')
+    result_3 = await ctx.call_activity(hello_retryable_act, retry_policy=retry_policy)
+    print(f'Activity 3 returned {result_3}')
+    result_4 = await ctx.call_child_workflow(child_retryable_wf, retry_policy=retry_policy)
+    print(f'Child workflow returned {result_4}')
+
+    # Event vs timeout using when_any
+    first = await ctx.when_any(
+        [
+            ctx.wait_for_external_event(event_name),
+            ctx.create_timer(timedelta(seconds=30)),
+        ]
+    )
+
+    # Proceed only if event won
+    if isinstance(first, dict) and 'event' in first:
+        await ctx.call_activity(hello_act, input=100)
+        await ctx.call_activity(hello_act, input=1000)
+        return 'Completed'
+    return 'Timeout'
+
+
+@wfr.activity(name='async_hello_act')
+def hello_act(ctx: WorkflowActivityContext, wf_input):
+    global counter
+    counter += wf_input
+    return f'Activity returned {wf_input}'
+
+
+@wfr.activity(name='async_hello_retryable_act')
+def hello_retryable_act(ctx: WorkflowActivityContext):
+    global retry_count
+    if (retry_count % 2) == 0:
+        retry_count += 1
+        raise ValueError('Retryable Error')
+    retry_count += 1
+    return f'Activity returned {retry_count}'
+
+
+@wfr.async_workflow(name=child_workflow_name)
+async def child_retryable_wf(ctx: AsyncWorkflowContext):
+    # Call activity with retry and simulate retryable workflow failure until certain state
+    child_activity_result = await ctx.call_activity(
+        act_for_child_wf, input='x', retry_policy=retry_policy
+    )
+    print(f'Child activity returned {child_activity_result}')
+    # In a real sample, you might check state and raise to trigger retry
+    return 'ok'
+
+
+@wfr.activity(name='async_act_for_child_wf')
+def act_for_child_wf(ctx: WorkflowActivityContext, inp):
+    global child_orchestrator_string
+    child_orchestrator_string += inp
+
+
+def main():
+    wfr.start()
+    wf_client = DaprWorkflowClient()
+
+    wf_client.schedule_new_workflow(
+        workflow=hello_world_wf, input=input_data, instance_id=instance_id
+    )
+
+    wf_client.wait_for_workflow_start(instance_id)
+
+    # Let initial activities run
+    sleep(5)
+
+    # Raise event to continue
+    wf_client.raise_workflow_event(
+        instance_id=instance_id, event_name=event_name, data={'ok': True}
+    )
+
+    # Wait for completion
+    state = wf_client.wait_for_workflow_completion(instance_id, timeout_in_seconds=60)
+    print(f'Workflow status: {state.runtime_status.name}')
+
+    wfr.shutdown()
+
+
+if __name__ == '__main__':
+    main()