initial code for decorators and optional naming of workflows and activities

Signed-off-by: Mukundan Sundararajan <65565396+mukundansundar@users.noreply.github.com>

Author: mukundansundar

README.md

@@ -81,6 +81,7 @@ cd python-sdk
 pip3 install -e .
 pip3 install -e ./ext/dapr-ext-grpc/
 pip3 install -e ./ext/dapr-ext-fastapi/
+pip3 install -e ./ext/dapr-ext-workflow/
 ```
 
 3. Install required packages

examples/workflow/fan_out_fan_in.py

@@ -12,9 +12,13 @@
 
 import time
 from typing import List
+from dapr.ext.workflow import workflow, activity, WorkflowRuntime
 import dapr.ext.workflow as wf
 
+# TODO: Modify code such that implicit creation of default WorkflowRuntime is possible
+workflowRuntime = wf.WorkflowRuntime("localhost", "50001")
 
+@workflow(name="batch_processing")
 def batch_processing_workflow(ctx: wf.DaprWorkflowContext, wf_input: int):
     # get a batch of N work items to process in parallel
     work_batch = yield ctx.call_activity(get_work_batch, input=wf_input)
@@ -27,30 +31,25 @@ def batch_processing_workflow(ctx: wf.DaprWorkflowContext, wf_input: int):
     total = sum(outputs)
     yield ctx.call_activity(process_results, input=total)
 
-
+@activity(name="get_batch")
 def get_work_batch(ctx, batch_size: int) -> List[int]:
     return [i + 1 for i in range(batch_size)]
 
-
+@activity
 def process_work_item(ctx, work_item: int) -> int:
     print(f'Processing work item: {work_item}.')
     time.sleep(5)
     result = work_item * 2
     print(f'Work item {work_item} processed. Result: {result}.')
     return result
 
-
+@activity(name="final_process")
 def process_results(ctx, final_result: int):
     print(f'Final result: {final_result}.')
 
 
 if __name__ == '__main__':
-    workflowRuntime = wf.WorkflowRuntime("localhost", "50001")
-    workflowRuntime.register_workflow(batch_processing_workflow)
-    workflowRuntime.register_activity(get_work_batch)
-    workflowRuntime.register_activity(process_work_item)
-    workflowRuntime.register_activity(process_results)
-    workflowRuntime.start()
+    WorkflowRuntime.start()
 
     wf_client = wf.DaprWorkflowClient()
     instance_id = wf_client.schedule_new_workflow(
@@ -59,4 +58,4 @@ def process_results(ctx, final_result: int):
     print(f'Workflow started. Instance ID: {instance_id}')
     state = wf_client.wait_for_workflow_completion(instance_id)
 
-    workflowRuntime.shutdown()
+    WorkflowRuntime.shutdown()

examples/workflow/human_approval.py

@@ -16,8 +16,11 @@
 import time
 
 from dapr.clients import DaprClient
+from dapr.ext.workflow import workflow, activity, WorkflowRuntime
 import dapr.ext.workflow as wf
 
+# TODO: Modify code such that implicit creation of default WorkflowRuntime is possible
+workflowRuntime = wf.WorkflowRuntime("localhost", "50001")
 
 @dataclass
 class Order:
@@ -37,7 +40,7 @@ class Approval:
     def from_dict(dict):
         return Approval(**dict)
 
-
+@workflow(name="purchase_order_wf")
 def purchase_order_workflow(ctx: wf.DaprWorkflowContext, order: Order):
     # Orders under $1000 are auto-approved
     if order.cost < 1000:
@@ -59,10 +62,12 @@ def purchase_order_workflow(ctx: wf.DaprWorkflowContext, order: Order):
     return f"Approved by '{approval_details.approver}'"
 
 
+@activity(name="send_approval")
 def send_approval_request(_, order: Order) -> None:
     print(f'*** Requesting approval from user for order: {order}')
 
 
+@activity
 def place_order(_, order: Order) -> None:
     print(f'*** Placing order: {order}')
 
@@ -76,12 +81,8 @@ def place_order(_, order: Order) -> None:
     parser.add_argument("--timeout", type=int, default=60, help="Timeout in seconds")
     args = parser.parse_args()
 
-    # configure and start the workflow runtime
-    workflowRuntime = wf.WorkflowRuntime("localhost", "50001")
-    workflowRuntime.register_workflow(purchase_order_workflow)
-    workflowRuntime.register_activity(send_approval_request)
-    workflowRuntime.register_activity(place_order)
-    workflowRuntime.start()
+    # start the workflow runtime
+    WorkflowRuntime.start()
 
     # Start a purchase order workflow using the user input
     order = Order(args.cost, "MyProduct", 1)
@@ -119,4 +120,4 @@ def prompt_for_approval():
     except TimeoutError:
         print("*** Workflow timed out!")
 
-    workflowRuntime.shutdown()
+    WorkflowRuntime.shutdown()

examples/workflow/monitor.py

@@ -13,15 +13,19 @@
 from dataclasses import dataclass
 from datetime import timedelta
 import random
+from dapr.ext.workflow import workflow, activity, WorkflowRuntime
 import dapr.ext.workflow as wf
 
+# TODO: Modify code such that implicit creation of default WorkflowRuntime is possible
+workflowRuntime = wf.WorkflowRuntime()
 
 @dataclass
 class JobStatus:
     job_id: str
     is_healthy: bool
 
 
+@workflow(name="status_monitor")
 def status_monitor_workflow(ctx: wf.DaprWorkflowContext, job: JobStatus):
     # poll a status endpoint associated with this job
     status = yield ctx.call_activity(check_status, input=job)
@@ -43,20 +47,18 @@ def status_monitor_workflow(ctx: wf.DaprWorkflowContext, job: JobStatus):
     ctx.continue_as_new(job)
 
 
+@activity
 def check_status(ctx, _) -> str:
     return random.choice(["healthy", "unhealthy"])
 
 
+@activity
 def send_alert(ctx, message: str):
     print(f'*** Alert: {message}')
 
 
 if __name__ == '__main__':
-    workflowRuntime = wf.WorkflowRuntime()
-    workflowRuntime.register_workflow(status_monitor_workflow)
-    workflowRuntime.register_activity(check_status)
-    workflowRuntime.register_activity(send_alert)
-    workflowRuntime.start()
+    WorkflowRuntime.start()
 
     wf_client = wf.DaprWorkflowClient()
     job_id = "job1"
@@ -75,4 +77,4 @@ def send_alert(ctx, message: str):
         print(f'Workflow already running. Instance ID: {job_id}')
 
     input("Press Enter to stop...\n")
-    workflowRuntime.shutdown()
+    WorkflowRuntime.shutdown()

examples/workflow/task_chaining.py

@@ -10,9 +10,14 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+from dapr.ext.workflow import workflow, activity, WorkflowRuntime
 import dapr.ext.workflow as wf
 
 
+wfr = WorkflowRuntime("localhost", "50001")
+
+
+@workflow(name="random_workflow")
 def task_chain_workflow(ctx: wf.DaprWorkflowContext, wf_input: int):
     try:
         result1 = yield ctx.call_activity(step1, input=wf_input)
@@ -24,38 +29,35 @@ def task_chain_workflow(ctx: wf.DaprWorkflowContext, wf_input: int):
     return [result1, result2, result3]
 
 
+@activity()
 def step1(ctx, activity_input):
     print(f'Step 1: Received input: {activity_input}.')
     # Do some work
     return activity_input + 1
 
 
+@activity()
 def step2(ctx, activity_input):
     print(f'Step 2: Received input: {activity_input}.')
     # Do some work
     return activity_input * 2
 
 
+@activity()
 def step3(ctx, activity_input):
     print(f'Step 3: Received input: {activity_input}.')
     # Do some work
     return activity_input ^ 2
 
 
+@activity()
 def error_handler(ctx, error):
     print(f'Executing error handler: {error}.')
     # Do some compensating work
 
 
 if __name__ == '__main__':
-    workflowRuntime = wf.WorkflowRuntime("localhost", "50001")
-    workflowRuntime.register_workflow(task_chain_workflow)
-    workflowRuntime.register_activity(step1)
-    workflowRuntime.register_activity(step2)
-    workflowRuntime.register_activity(step3)
-    workflowRuntime.register_activity(error_handler)
-    workflowRuntime.start()
-
+    WorkflowRuntime.start()
     wf_client = wf.DaprWorkflowClient()
     instance_id = wf_client.schedule_new_workflow(
         workflow=task_chain_workflow,
@@ -64,4 +66,4 @@ def error_handler(ctx, error):
     state = wf_client.wait_for_workflow_completion(instance_id)
     print(f'Workflow completed! Status: {state.runtime_status}')
 
-    workflowRuntime.shutdown()
+    WorkflowRuntime.shutdown()

ext/dapr-ext-workflow/dapr/ext/workflow/__init__.py

@@ -14,7 +14,7 @@
 """
 
 # Import your main classes here
-from dapr.ext.workflow.workflow_runtime import WorkflowRuntime
+from dapr.ext.workflow.workflow_runtime import WorkflowRuntime, workflow, activity
 from dapr.ext.workflow.dapr_workflow_client import DaprWorkflowClient
 from dapr.ext.workflow.dapr_workflow_context import DaprWorkflowContext, when_all, when_any
 from dapr.ext.workflow.workflow_activity_context import WorkflowActivityContext
@@ -28,5 +28,7 @@
     'WorkflowState',
     'WorkflowStatus',
     'when_all',
-    'when_any'
+    'when_any',
+    'workflow',
+    'activity'
 ]

ext/dapr-ext-workflow/dapr/ext/workflow/dapr_workflow_client.py

@@ -75,8 +75,15 @@ def schedule_new_workflow(self, workflow: Workflow, *, input: Optional[TInput] =
         Returns:
             The ID of the scheduled workflow instance.
         """
-        return self.__obj.schedule_new_orchestration(workflow.__name__, input=input,
-                                                     instance_id=instance_id, start_at=start_at)
+        try:
+            workflow.__getattribute__('_registered_name')
+            return self.__obj.schedule_new_orchestration(workflow._registered_name, input=input,
+                                                         instance_id=instance_id,
+                                                         start_at=start_at)
+        except AttributeError:
+            return self.__obj.schedule_new_orchestration(workflow.__name__, input=input,
+                                                         instance_id=instance_id,
+                                                         start_at=start_at)
 
     def get_workflow_state(self, instance_id: str, *,
                            fetch_payloads: bool = True) -> Optional[WorkflowState]:

ext/dapr-ext-workflow/dapr/ext/workflow/dapr_workflow_context.py

@@ -53,7 +53,10 @@ def create_timer(self, fire_at: Union[datetime, timedelta]) -> task.Task:
 
     def call_activity(self, activity: Callable[[WorkflowActivityContext, TInput], TOutput], *,
                       input: TInput = None) -> task.Task[TOutput]:
-        return self.__obj.call_activity(activity=activity.__name__, input=input)
+        try:
+            return self.__obj.call_activity(activity=activity._registered_name, input=input)
+        except AttributeError:
+            return self.__obj.call_activity(activity=activity.__name__, input=input)
 
     def call_child_workflow(self, workflow: Workflow, *,
                             input: Optional[TInput],
@@ -62,7 +65,12 @@ def wf(ctx: task.OrchestrationContext, inp: TInput):
             daprWfContext = DaprWorkflowContext(ctx)
             return workflow(daprWfContext, inp)
         # copy workflow name so durabletask.worker can find the orchestrator in its registry
-        wf.__name__ = workflow.__name__
+
+        # Any workflow function using python decorator will have a _registered_name attribute
+        try:
+            wf.__name__ = workflow._registered_name
+        except AttributeError:
+            wf.__name__ = workflow.__name__
         return self.__obj.call_sub_orchestrator(wf, input=input, instance_id=instance_id)
 
     def wait_for_external_event(self, name: str) -> task.Task: