feat: define workflow signal decorator in workflowDefinition

Signed-off-by: Tim Li <[email protected]>

Author: timl3136

cadence/workflow.py

@@ -59,10 +59,16 @@ class WorkflowDefinition:
     Provides type safety and metadata for workflow classes.
     """
 
-    def __init__(self, cls: Type, name: str, run_method_name: str):
+    def __init__(self, cls: Type, name: str, run_method_name: str, signals: dict[str, Callable[..., Any]]):
         self._cls = cls
         self._name = name
         self._run_method_name = run_method_name
+        self._signals = signals
+
+    @property
+    def signals(self) -> dict[str, Callable[..., Any]]:
+        """Get the signals."""
+        return self._signals
 
     @property
     def name(self) -> str:
@@ -98,6 +104,9 @@ def wrap(cls: Type, opts: WorkflowDefinitionOptions) -> "WorkflowDefinition":
             name = opts["name"]
 
         # Validate that the class has exactly one run method and find it
+        # Also validate that class does not have multiple signal methods with the same name
+        signals: dict[str, Callable[..., Any]] = {}
+        signal_names: dict[str, str] = {}  # Map signal name to method name for duplicate detection
         run_method_name = None
         for attr_name in dir(cls):
             if attr_name.startswith("_"):
@@ -114,11 +123,22 @@ def wrap(cls: Type, opts: WorkflowDefinitionOptions) -> "WorkflowDefinition":
                         f"Multiple @workflow.run methods found in class {cls.__name__}"
                     )
                 run_method_name = attr_name
+            
+            if hasattr(attr, "_workflow_signal"):
+                signal_name = getattr(attr, "_workflow_signal")
+                if signal_name in signal_names:
+                    raise ValueError(
+                        f"Multiple @workflow.signal methods found in class {cls.__name__} "
+                        f"with signal name '{signal_name}': '{attr_name}' and '{signal_names[signal_name]}'"
+                    )
+                signals[attr_name] = attr
+                signal_names[signal_name] = attr_name
 
         if run_method_name is None:
             raise ValueError(f"No @workflow.run method found in class {cls.__name__}")
 
-        return WorkflowDefinition(cls, name, run_method_name)
+        return WorkflowDefinition(cls, name, run_method_name, signals)
+
 
 
 def run(func: Optional[T] = None) -> Union[T, Callable[[T], T]]:
@@ -161,6 +181,33 @@ def decorator(f: T) -> T:
         # Called without parentheses: @workflow.run
         return decorator(func)
 
+def signal(name: str | None = None) -> Callable[[T], T]:
+    """
+    Decorator to mark a method as a workflow signal handler.
+
+    Example:
+       	@workflow.signal(name="approval_channel")
+        async def approve(self, approved: bool):
+            self.approved = approved
+
+    Args:
+        name: The name of the signal
+
+    Returns:
+        The decorated method with workflow signal metadata
+
+    Raises:
+        ValueError: If name is not provided
+
+    """
+    if name is None:
+        raise ValueError("name is required")
+
+    def decorator(f: T) -> T:
+        f._workflow_signal = name  # type: ignore
+        return f    
+    # Only allow @workflow.signal(name), require name to be explicitly provided
+    return decorator
 
 @dataclass
 class WorkflowInfo:

tests/cadence/worker/test_registry.py

@@ -212,3 +212,89 @@ async def run(self, input: str) -> str:
         workflow_def = reg.get_workflow("custom_workflow_name")
         assert workflow_def.name == "custom_workflow_name"
         assert workflow_def.cls == CustomWorkflow
+
+    def test_workflow_with_signal(self):
+        """Test workflow with signal handler."""
+        reg = Registry()
+
+        @reg.workflow
+        class WorkflowWithSignal:
+            @workflow.run
+            async def run(self):
+                return "done"
+
+            @workflow.signal(name="approval")
+            async def handle_approval(self, approved: bool):
+                self.approved = approved
+
+        workflow_def = reg.get_workflow("WorkflowWithSignal")
+        assert isinstance(workflow_def, WorkflowDefinition)
+        assert len(workflow_def.signals) == 1
+        assert "handle_approval" in workflow_def.signals
+        assert hasattr(workflow_def.signals["handle_approval"], "_workflow_signal")
+        assert workflow_def.signals["handle_approval"]._workflow_signal == "approval"
+
+    def test_workflow_with_multiple_signals(self):
+        """Test workflow with multiple signal handlers."""
+        reg = Registry()
+
+        @reg.workflow
+        class WorkflowWithMultipleSignals:
+            @workflow.run
+            async def run(self):
+                return "done"
+
+            @workflow.signal(name="approval")
+            async def handle_approval(self, approved: bool):
+                self.approved = approved
+
+            @workflow.signal(name="cancel")
+            async def handle_cancel(self):
+                self.cancelled = True
+
+        workflow_def = reg.get_workflow("WorkflowWithMultipleSignals")
+        assert len(workflow_def.signals) == 2
+        assert "handle_approval" in workflow_def.signals
+        assert "handle_cancel" in workflow_def.signals
+        assert getattr(workflow_def.signals["handle_approval"], "_workflow_signal") == "approval"
+        assert getattr(workflow_def.signals["handle_cancel"], "_workflow_signal") == "cancel"
+
+    def test_signal_decorator_requires_name(self):
+        """Test that signal decorator requires name parameter."""
+        with pytest.raises(ValueError, match="name is required"):
+            @workflow.signal()
+            async def test_signal(self):
+                pass
+
+    def test_workflow_without_signals(self):
+        """Test that workflow without signals has empty signals dict."""
+        reg = Registry()
+
+        @reg.workflow
+        class WorkflowWithoutSignals:
+            @workflow.run
+            async def run(self):
+                return "done"
+
+        workflow_def = reg.get_workflow("WorkflowWithoutSignals")
+        assert isinstance(workflow_def.signals, dict)
+        assert len(workflow_def.signals) == 0
+
+    def test_duplicate_signal_names_error(self):
+        """Test that duplicate signal names raise ValueError."""
+        reg = Registry()
+
+        with pytest.raises(ValueError, match="Multiple.*signal.*found.*with signal name 'approval'"):
+            @reg.workflow
+            class WorkflowWithDuplicateSignalNames:
+                @workflow.run
+                async def run(self):
+                    return "done"
+
+                @workflow.signal(name="approval")
+                async def handle_approval(self, approved: bool):
+                    self.approved = approved
+
+                @workflow.signal(name="approval")
+                async def handle_approval_different(self):
+                    self.also_approved = True