diff --git a/dapr/ext/workflow/AGENTS.md b/dapr/ext/workflow/AGENTS.md index 80ac07d7b..4874a576c 100644 --- a/dapr/ext/workflow/AGENTS.md +++ b/dapr/ext/workflow/AGENTS.md @@ -131,7 +131,7 @@ Workflow (orchestrator) functions must remain generators (`def` with `yield`). T Client for workflow lifecycle management: -- `schedule_new_workflow(workflow, *, input, instance_id, start_at, reuse_id_policy)` → returns `instance_id` +- `schedule_new_workflow(workflow, *, input, instance_id, start_at, reuse_id_policy)` → returns `instance_id`. `reuse_id_policy` is deprecated (no effect, emits `DeprecationWarning`): an instance ID can always be reused once the existing instance is in a terminal state. - `get_workflow_state(instance_id, *, fetch_payloads=True)` → `Optional[WorkflowState]` - `wait_for_workflow_start(instance_id, *, fetch_payloads, timeout_in_seconds)` - `wait_for_workflow_completion(instance_id, *, fetch_payloads, timeout_in_seconds)` diff --git a/dapr/ext/workflow/_durabletask/client.py b/dapr/ext/workflow/_durabletask/client.py index dfc507694..86c54fd47 100644 --- a/dapr/ext/workflow/_durabletask/client.py +++ b/dapr/ext/workflow/_durabletask/client.py @@ -62,8 +62,15 @@ def __str__(self): return helpers.get_orchestration_status_str(self.value) +# TODO(v1.21): remove OrchestrationIdReuseAction, WorkflowIdReusePolicy, and the reuse_id_policy +# parameter on schedule_new_orchestration (sync and aio) — deprecated in 1.18, kept for the +# 3-version policy. class OrchestrationIdReuseAction(Enum): - """Action to take when scheduling an orchestration whose ID already exists.""" + """Action to take when scheduling an orchestration whose ID already exists. + + Deprecated: has no effect and will be removed in a future release. An orchestration ID + can always be reused once the existing instance has reached a terminal state. + """ ERROR = 0 IGNORE = 1 @@ -72,7 +79,11 @@ class OrchestrationIdReuseAction(Enum): @dataclass class WorkflowIdReusePolicy: - """Policy controlling what happens when a new orchestration is scheduled with an ID that already exists.""" + """Policy controlling what happens when a new orchestration is scheduled with an ID that already exists. + + Deprecated: has no effect and will be removed in a future release. An orchestration ID + can always be reused once the existing instance has reached a terminal state. + """ action: OrchestrationIdReuseAction operation_status: list[OrchestrationStatus] diff --git a/dapr/ext/workflow/aio/dapr_workflow_client.py b/dapr/ext/workflow/aio/dapr_workflow_client.py index c8001db98..8a6072cf2 100644 --- a/dapr/ext/workflow/aio/dapr_workflow_client.py +++ b/dapr/ext/workflow/aio/dapr_workflow_client.py @@ -17,6 +17,7 @@ from datetime import datetime from typing import Any, Optional, TypeVar, Union +from warnings import warn from grpc.aio import AioRpcError @@ -109,12 +110,22 @@ async def schedule_new_workflow( start_at: The time when the workflow instance should start executing. If not specified or if a date-time in the past is specified, the workflow instance will be scheduled immediately. - reuse_id_policy: Optional policy to reuse the workflow id when there is a conflict with - an existing workflow instance. + reuse_id_policy: Deprecated and has no effect; it will be removed in a future + release. A workflow instance ID can always be reused once the existing instance + with that ID has reached a terminal state (e.g. COMPLETED, FAILED, or TERMINATED). Returns: The ID of the scheduled workflow instance. """ + # TODO(v1.21): remove reuse_id_policy — deprecated in 1.18, kept for the 3-version policy. + if reuse_id_policy is not None: + warn( + 'reuse_id_policy is deprecated and has no effect; it will be removed in a ' + 'future release. A workflow instance ID can always be reused once the existing ' + 'instance with that ID has reached a terminal state.', + DeprecationWarning, + stacklevel=2, + ) if isinstance(workflow, str): workflow_name = workflow elif hasattr(workflow, '_dapr_alternate_name'): diff --git a/dapr/ext/workflow/dapr_workflow_client.py b/dapr/ext/workflow/dapr_workflow_client.py index c72adef12..dca65c200 100644 --- a/dapr/ext/workflow/dapr_workflow_client.py +++ b/dapr/ext/workflow/dapr_workflow_client.py @@ -17,6 +17,7 @@ from datetime import datetime from typing import Any, Optional, TypeVar, Union +from warnings import warn from grpc import RpcError @@ -111,12 +112,22 @@ def schedule_new_workflow( start_at: The time when the workflow instance should start executing. If not specified or if a date-time in the past is specified, the workflow instance will be scheduled immediately. - reuse_id_policy: Optional policy to reuse the workflow id when there is a conflict with - an existing workflow instance. + reuse_id_policy: Deprecated and has no effect; it will be removed in a future + release. A workflow instance ID can always be reused once the existing instance + with that ID has reached a terminal state (e.g. COMPLETED, FAILED, or TERMINATED). Returns: The ID of the scheduled workflow instance. """ + # TODO(v1.21): remove reuse_id_policy — deprecated in 1.18, kept for the 3-version policy. + if reuse_id_policy is not None: + warn( + 'reuse_id_policy is deprecated and has no effect; it will be removed in a ' + 'future release. A workflow instance ID can always be reused once the existing ' + 'instance with that ID has reached a terminal state.', + DeprecationWarning, + stacklevel=2, + ) if isinstance(workflow, str): workflow_name = workflow elif hasattr(workflow, '_dapr_alternate_name'): diff --git a/examples/workflow/monitor.py b/examples/workflow/monitor.py index d4f534df5..87ab0c6e3 100644 --- a/examples/workflow/monitor.py +++ b/examples/workflow/monitor.py @@ -71,7 +71,8 @@ def send_alert(ctx, message: str): pass if not status or status.runtime_status.name != 'RUNNING': - # TODO update to use reuse_id_policy + # Scheduling over a RUNNING instance fails; reuse of the ID is only allowed once the + # existing instance has reached a terminal state, hence the pre-check above. instance_id = wf_client.schedule_new_workflow( workflow=status_monitor_workflow, input=JobStatus(job_id=job_id, is_healthy=True), diff --git a/tests/clients/grpc_private.key b/tests/clients/grpc_private.key deleted file mode 100644 index 4664cb9b4..000000000 --- a/tests/clients/grpc_private.key +++ /dev/null @@ -1,52 +0,0 @@ ------BEGIN PRIVATE KEY----- -MIIJQwIBADANBgkqhkiG9w0BAQEFAASCCS0wggkpAgEAAoICAQCTBLvfzv88vPbc -n/IGzsdkTG099VTQjQkGQRG1d4MMFyk1tMRT40xYn4pEeqN8bN3Luy1NHiHqCTbR -+WWr+k1cM/q1H4LakTNSvMgcB9AdMTLKY4puEYABeYipOD/KNQl1dexBnDc65dCI -+fcQ0H+Qa5l9r7SYGHswQD+TqaIIH14jS4H5adhWxXxqM0L1BzwZ4iEHHxvp+w0w -P4xzeB1m+quwFboybcCXgOIlYa5Ro5JK7fkwht2WRqED3ebBwfolV6t1QTCp83c8 -c/t1RbqYKuzqd8Jc0Hb4FkV+IeZNgHhmWDePr2neiH5DBib0zKS1gFYWkkuXEUVc -S7jj1DlmC7mGOAtn0IPtcic9JITT+glTdSpaSgx8R0BXmdfJRtHUTda8GeA+IeBm -LRLRP/DSbqGCE4enXdDJx4rwaMv1jJygF/tz/0sH8qvxYlnbtD6nAfTJ+/G93g79 -OZnyzE0n02DDiBI9Dt1P9C4RmkIWFOEztS9Eq3VDO2sfMAgg0JznqMv1Y+fzRDG7 -/QFxBtEEj5bvRVuS6mBsPzNsY9HvVlTNFd0roa5eOtx0AbWr4hS9Ta8JE85f7Oc/ -T3Zr2ToRj1RPH1iWmYXsrnbvAonrBnsQ2v3skDFDb0vQNs0Jzg3w1VvUYSHwytZI -JjqmO5Ih4kyBfId9ebDUxeB0S/v2zwIDAQABAoICAAmTRCektBvHZ7ShRJwlssRy -wQxJNdSKoWajRxbKL+JrtI1nUDO+f0x0wTnW9JFRoQ7epAN8moZ2dZGVBez+vAC5 -vSYEZWHJSM4MtmPM+asARBr8z+YXUbHG5QKDqichvGBgGJq1V/Hue3e2zN0IsPMK -ojaWnDm27nik1pjOPyiURaDuDz5CIvFgopfJuusTWFpYz5ngVtF7pM8Np7iL8pSw -b6b9fOEOV0rTl01wxtYeXZRvwP9+HGFOb4VwjTgoloHNuFpGSRGCMxpG24vQMAHI -We6mpvJwli8JerRnKAUumg9Gxs4FUlN7qrpZ/jnxBV4LSJAaLlfEEjM1Forc8z5i -IxGHv6M7I2nrYaSybk2LbsQ4gvoDpiHrTO5HK/r25iShMFoHCsvSZ+A57fXp/IAx -mij7UQzpdVwU7lqcVIYoJCxc1SPZX32howm0l+t4IKFYjd+oz1NxTyOZuFMdYWcB -tVrqGedwvFDnTtacxLakoLt5P/QztnpTis1YP9lagE6NlP8hPiMqmXjNV5nsfFiM -oVkuckCms57ZF6IyKzEKzPmi0BJ2O+mwCIUDXM548V4hEgDf6yoMc0pr9vSdBLDh -D5otiF1XNjyeC/2oS4TZO21lirgmNfafp7MOEjbojqbwmzVxK5kp0+SZ/mCSNFiU -sD3vLExHukoO+zDp6GuBAoIBAQDE1yNzDokRpH5WsX/L4n4KIkiwJWOkn6pBz4u4 -8m7yjwIY+3Wo5D4C2zvhU3K/RR6qh7wIXW0sLbNskz68b2SdDoz3SrDR6sMF5Hxd -d38Nw6Xd0NNWTPq0ILrVxXEsrcxFPwGNi1GebznGBa5ONk5oQJsegpqVivUefMz1 -SjrCHkCT+mOXOATaYZOWyOOjbdvSG7trrThLQa+2dbihI8ETGj/vblLW0K/7UCsv -9y5pL/phWjUUhr3+g7cfvxkC0NWj9HlMF7XlRCZEla+Xckq/FkwoPcB10y7vxFts -F3WEJv+FX4YcotkvxIRBKhyopaCvwUqWHXYRF62U2uzy3F4vAoIBAQC/NE75+/Kc -aHmROvV0eY3BGlR8qjmKnoGRZZH73pRVM8Jv3Jf4+9QlnJPpo+K9dgyYg14gDvwT -m+TSb9c5mJvJlSVGph4qXm8IO1hx3tTi+loxpVSA/9aU8LlIOMiUlHeCWt165dQh -WUjk7lYT+SaeiNbwuGFqRa6BMMTdIdeFgbtuhW4EB1nq3cvR79udf+KncLrnXY+b -VllQ0JnpLGqxzHtcsUV5Uxf5/OFCrVKwS/OAS5hcsg4mfl9scuSILC9dCQlDo6Qn -daDT7aT51fG0aawxrRg1D9iFe7r6IkSw0prlY1ElzGqxxdWVGCgK6oOtqRGnf2/i -dWV9bXX2FWlhAoIBAQCzNPMvQyKDj3D+WKdy5ge+DiyDQecP79VACUa54O4loxFT -0UvzsajLTPGdWS30+WbL9ru4GebU+96IkR3izL+bRa4NTlDhLIzVhom7RpnD3EXc -RszBm1/8MngnO5SoCqsYuB4f0aFDTxwYLv/ccRC9VWRkSJ+YWmSDlze6chRAuu4s -Sx/0gqnYcFHAS7EED3a8OUdO5uuDiJAC2/Z1rmmPjKx8vcJplKFDel93UeFNHsMj -cs80mt/LIfvMyVxp9dsrY94y6rtNudflRSoSBZoLkq9PS7k3DEeFPw3EwS7aW+7y -bF9bHuAA3zdRZJem6vP6T04qkTP2p8n+5bon+X09AoIBAQCGf/BST1XxiliBvdyu -K1d5D+GAtGm3xhpUwgM4DvF/IeLTt3Q8FEG+l29A+iTkq6bengJXQNM3pAEJutzG -SkDEQ6MuGGq92SH3w29K4roUVJGfvLI98DntGYzsbe1C/1E2JF8KNaPMNbdeOv3Y -HPp6qL2yJbF5BT1k6X6CI20tH/5OF5Q1R0qMOatHqZduKmiw6k6/tAPx/ILlo2ex -K6p+Gd5pbNX3K2mBU87NjYjONQ88H3E1O5dSorDTY9q4o4RJyLJ4nyM5Erioe9IC -YOoew/r6Es4lfVkNROVZUUPKsst9MykBEtO26/h7YbvBOgiY34Zose9teLGE0rwY -+CThAoIBAHx/8aGgQU1mF+zwf0fCB1l+Hdp4oLrLInPW+RRu5JuGDa0dCsXWRjtF -DdFPTbjAUj6MzXmAxfQ7UEIjybaJJJgHFsMT0K40VgQBbnxZnouPB5dcEjhXWcX7 -+eUZAANSCOpVblnjxpFa4iCCyjWj8HmoUfMr6/qyAWppvIozqI+DOv1yWiXb9WcO -7yx2vblQzqksGx4ujZglPMq4bMtgePYcIpEBtY+u97ZXGB7IYCD7G9oe1ywUzWS6 -I6aRoNT/CFWouyxpCeiejff69aRgKGF4yU2PORrd6nRHxwreHr+9tDi9BVSCrmuo -9mDh5sVIS25V4SWsjqw5dbm/qdE6suA= ------END PRIVATE KEY----- diff --git a/tests/clients/grpc_selfsigned.pem b/tests/clients/grpc_selfsigned.pem deleted file mode 100644 index 5eb11e8ef..000000000 --- a/tests/clients/grpc_selfsigned.pem +++ /dev/null @@ -1,28 +0,0 @@ ------BEGIN CERTIFICATE----- -MIIE0jCCArqgAwIBAgIUG8GKBKK6pRPhbo++9jJ1MM/13g8wDQYJKoZIhvcNAQEN -BQAwIzENMAsGA1UECgwERGFwcjESMBAGA1UEAwwJbG9jYWxob3N0MB4XDTI2MDcw -ODA4MTMwMVoXDTI2MDcwOTA4MTMwMVowIzENMAsGA1UECgwERGFwcjESMBAGA1UE -AwwJbG9jYWxob3N0MIICIjANBgkqhkiG9w0BAQEFAAOCAg8AMIICCgKCAgEAkwS7 -387/PLz23J/yBs7HZExtPfVU0I0JBkERtXeDDBcpNbTEU+NMWJ+KRHqjfGzdy7st -TR4h6gk20fllq/pNXDP6tR+C2pEzUrzIHAfQHTEyymOKbhGAAXmIqTg/yjUJdXXs -QZw3OuXQiPn3ENB/kGuZfa+0mBh7MEA/k6miCB9eI0uB+WnYVsV8ajNC9Qc8GeIh -Bx8b6fsNMD+Mc3gdZvqrsBW6Mm3Al4DiJWGuUaOSSu35MIbdlkahA93mwcH6JVer -dUEwqfN3PHP7dUW6mCrs6nfCXNB2+BZFfiHmTYB4Zlg3j69p3oh+QwYm9MyktYBW -FpJLlxFFXEu449Q5Zgu5hjgLZ9CD7XInPSSE0/oJU3UqWkoMfEdAV5nXyUbR1E3W -vBngPiHgZi0S0T/w0m6hghOHp13QyceK8GjL9YycoBf7c/9LB/Kr8WJZ27Q+pwH0 -yfvxvd4O/TmZ8sxNJ9Ngw4gSPQ7dT/QuEZpCFhThM7UvRKt1QztrHzAIINCc56jL -9WPn80Qxu/0BcQbRBI+W70VbkupgbD8zbGPR71ZUzRXdK6GuXjrcdAG1q+IUvU2v -CRPOX+znP092a9k6EY9UTx9YlpmF7K527wKJ6wZ7ENr97JAxQ29L0DbNCc4N8NVb -1GEh8MrWSCY6pjuSIeJMgXyHfXmw1MXgdEv79s8CAwEAATANBgkqhkiG9w0BAQ0F -AAOCAgEAPt8Q4eOhmeTpjPqSHh18ow+Bk18rWhoZ3iMMrilf0ceewY3z2wHLwLAX -jCUOoHAeqCq5+Sb/qA65Uh9/9uAuZz5hkZkSqa42Y5SM907yzmG3oSN4l4E6b8vp -JQQq+qQ00DFXLad1mqHmvmKTMr492j2/WaA9bO92IJ5yq0074c2O4VJlyiqCHuec -sIOYOi0J+0ZmFAR5YziLsMz9Gqpk486s4Y2+WE90zaAQ5l4khmnFz35DHJDMAuuE -AzrDaWsepBvS7OFDu6ONInc9FYAUOilk+/vPPG3ajOJRgkwnB8D8LEvfrmck2mnH -W4g2iXsAzheuFH6v4uvvcuPCUi5rrgBGGARCZbN0imx0qrIM8rbqiQAflvDjUlff -GXnOz9O5VIxKraB0tn5dxe+n0+CQhkOK2rilzZld58YlZ5G9uvvS75y3sipp088K -rmSLpyo2npBS+RUPTwpK8nKB69t8xgxqmrwsawEI2gT8n+oU6IcHj0+tTxUL2mtm -3YMlez8Kh0srX2h72Ksamq4n2AOV0iMEt+/7rmd8Oh5RC7v6LY1Is5b3MO3iaNZH -PHxDT2lIxdNEO8oB+HOHVAN/aXgXmwW3OOAQyCMONNiMEgs2mLUzmZ8/cFd/02H9 -7rI9Bs4BreK5C4VY1EnSmV6nSQ3kn9H+/vK3vNpOc1vvZRWaEIs= ------END CERTIFICATE----- diff --git a/tests/ext/workflow/test_workflow_client.py b/tests/ext/workflow/test_workflow_client.py index fa2796a85..257e72b35 100644 --- a/tests/ext/workflow/test_workflow_client.py +++ b/tests/ext/workflow/test_workflow_client.py @@ -14,6 +14,7 @@ """ import unittest +import warnings from datetime import datetime from typing import Any, Union from unittest import mock @@ -198,6 +199,33 @@ def test_schedule_workflow_by_name_string(self): assert result == mock_schedule_result assert fake_client.last_scheduled_workflow_name == 'my_registered_workflow' + def test_schedule_workflow_reuse_id_policy_is_deprecated(self): + fake_client = FakeTaskHubGrpcClient() + with mock.patch( + 'dapr.ext.workflow._durabletask.client.TaskHubGrpcClient', return_value=fake_client + ): + wfClient = DaprWorkflowClient() + reuse_id_policy = client.WorkflowIdReusePolicy( + action=client.OrchestrationIdReuseAction.TERMINATE, + operation_status=[client.OrchestrationStatus.RUNNING], + ) + with self.assertWarns(DeprecationWarning): + result = wfClient.schedule_new_workflow( + workflow='my_registered_workflow', reuse_id_policy=reuse_id_policy + ) + assert result == mock_schedule_result + + def test_schedule_workflow_without_reuse_id_policy_does_not_warn(self): + fake_client = FakeTaskHubGrpcClient() + with mock.patch( + 'dapr.ext.workflow._durabletask.client.TaskHubGrpcClient', return_value=fake_client + ): + wfClient = DaprWorkflowClient() + with warnings.catch_warnings(): + warnings.simplefilter('error', DeprecationWarning) + result = wfClient.schedule_new_workflow(workflow='my_registered_workflow') + assert result == mock_schedule_result + def test_client_functions(self): with mock.patch( 'dapr.ext.workflow._durabletask.client.TaskHubGrpcClient', diff --git a/tests/ext/workflow/test_workflow_client_aio.py b/tests/ext/workflow/test_workflow_client_aio.py index 63a653c93..6bdf2d5be 100644 --- a/tests/ext/workflow/test_workflow_client_aio.py +++ b/tests/ext/workflow/test_workflow_client_aio.py @@ -14,6 +14,7 @@ """ import unittest +import warnings from datetime import datetime from typing import Any, Union from unittest import mock @@ -202,6 +203,35 @@ async def test_schedule_workflow_by_name_string(self): assert result == mock_schedule_result assert fake_client.last_scheduled_workflow_name == 'my_registered_workflow' + async def test_schedule_workflow_reuse_id_policy_is_deprecated(self): + fake_client = FakeAsyncTaskHubGrpcClient() + with mock.patch( + 'dapr.ext.workflow._durabletask.aio.client.AsyncTaskHubGrpcClient', + return_value=fake_client, + ): + wfClient = DaprWorkflowClient() + reuse_id_policy = client.WorkflowIdReusePolicy( + action=client.OrchestrationIdReuseAction.TERMINATE, + operation_status=[client.OrchestrationStatus.RUNNING], + ) + with self.assertWarns(DeprecationWarning): + result = await wfClient.schedule_new_workflow( + workflow='my_registered_workflow', reuse_id_policy=reuse_id_policy + ) + assert result == mock_schedule_result + + async def test_schedule_workflow_without_reuse_id_policy_does_not_warn(self): + fake_client = FakeAsyncTaskHubGrpcClient() + with mock.patch( + 'dapr.ext.workflow._durabletask.aio.client.AsyncTaskHubGrpcClient', + return_value=fake_client, + ): + wfClient = DaprWorkflowClient() + with warnings.catch_warnings(): + warnings.simplefilter('error', DeprecationWarning) + result = await wfClient.schedule_new_workflow(workflow='my_registered_workflow') + assert result == mock_schedule_result + async def test_client_functions(self): with mock.patch( 'dapr.ext.workflow._durabletask.aio.client.AsyncTaskHubGrpcClient',