Dynamic workflows, activities, signals, and queries (#346)

Fixes #242

Author: cretz

.github/workflows/ci.yml

@@ -24,6 +24,12 @@ jobs:
           - os: ubuntu-arm
             runsOn: buildjet-4vcpu-ubuntu-2204-arm
     runs-on: ${{ matrix.runsOn || matrix.os }}
+    # For Windows there is currently a bug with Windows + pytest + warnings +
+    # importlib + Python < 3.12. Based on others' investigations, disabling
+    # bytecode fixes it.
+    # See https://github.com/temporalio/sdk-python/pull/346#issuecomment-1636108747
+    env:
+      PYTHONDONTWRITEBYTECODE: "${{ matrix.os == 'windows-latest' && '1' || '' }}"
     steps:
       - uses: actions/checkout@v2
         with:

README.md

@@ -539,6 +539,9 @@ Here are the decorators that can be applied:
 * `@workflow.defn` - Defines a workflow class
   * Must be defined on the class given to the worker (ignored if present on a base class)
   * Can have a `name` param to customize the workflow name, otherwise it defaults to the unqualified class name
+  * Can have `dynamic=True` which means all otherwise unhandled workflows fall through to this. If present, cannot have
+    `name` argument, and run method must accept a single parameter of `Sequence[temporalio.common.RawValue]` type. The
+    payload of the raw value can be converted via `workflow.payload_converter().from_payload`.
 * `@workflow.run` - Defines the primary workflow run method
   * Must be defined on the same class as `@workflow.defn`, not a base class (but can _also_ be defined on the same
     method of a base class)
@@ -553,7 +556,8 @@ Here are the decorators that can be applied:
   * The method's arguments are the signal's arguments
   * Can have a `name` param to customize the signal name, otherwise it defaults to the unqualified method name
   * Can have `dynamic=True` which means all otherwise unhandled signals fall through to this. If present, cannot have
-    `name` argument, and method parameters must be `self`, a string signal name, and a `*args` varargs param.
+    `name` argument, and method parameters must be `self`, a string signal name, and a
+    `Sequence[temporalio.common.RawValue]`.
   * Non-dynamic method can only have positional arguments. Best practice is to only take a single argument that is an
     object/dataclass of fields that can be added to as needed.
   * Return value is ignored
@@ -1051,6 +1055,10 @@ Some things to note about activity definitions:
   activity may need (e.g. a DB connection). The instance method should be what is registered with the worker.
 * Activities can also be defined on callable classes (i.e. classes with `__call__`). An instance of the class should be
   what is registered with the worker.
+* The `@activity.defn` can have `dynamic=True` set which means all otherwise unhandled activities fall through to this.
+  If present, cannot have `name` argument, and the activity function must accept a single parameter of
+  `Sequence[temporalio.common.RawValue]`. The payload of the raw value can be converted via
+  `activity.payload_converter().from_payload`.
 
 #### Types of Activities
 

temporalio/activity.py

@@ -35,6 +35,7 @@
 )
 
 import temporalio.common
+import temporalio.converter
 
 from .types import CallableType
 
@@ -51,11 +52,19 @@ def defn(
     ...
 
 
+@overload
+def defn(
+    *, no_thread_cancel_exception: bool = False, dynamic: bool = False
+) -> Callable[[CallableType], CallableType]:
+    ...
+
+
 def defn(
     fn: Optional[CallableType] = None,
     *,
     name: Optional[str] = None,
     no_thread_cancel_exception: bool = False,
+    dynamic: bool = False,
 ):
     """Decorator for activity functions.
 
@@ -64,15 +73,19 @@ def defn(
     Args:
         fn: The function to decorate.
         name: Name to use for the activity. Defaults to function ``__name__``.
+            This cannot be set if dynamic is set.
         no_thread_cancel_exception: If set to true, an exception will not be
             raised in synchronous, threaded activities upon cancellation.
+        dynamic: If true, this activity will be dynamic. Dynamic activities have
+            to accept a single 'Sequence[RawValue]' parameter. This cannot be
+            set to true if name is present.
     """
 
     def decorator(fn: CallableType) -> CallableType:
         # This performs validation
         _Definition._apply_to_callable(
             fn,
-            activity_name=name or fn.__name__,
+            activity_name=name or fn.__name__ if not dynamic else None,
             no_thread_cancel_exception=no_thread_cancel_exception,
         )
         return fn
@@ -132,7 +145,12 @@ class _Context:
     cancelled_event: _CompositeEvent
     worker_shutdown_event: _CompositeEvent
     shield_thread_cancel_exception: Optional[Callable[[], AbstractContextManager]]
+    payload_converter_class_or_instance: Union[
+        Type[temporalio.converter.PayloadConverter],
+        temporalio.converter.PayloadConverter,
+    ]
     _logger_details: Optional[Mapping[str, Any]] = None
+    _payload_converter: Optional[temporalio.converter.PayloadConverter] = None
 
     @staticmethod
     def current() -> _Context:
@@ -155,6 +173,18 @@ def logger_details(self) -> Mapping[str, Any]:
             self._logger_details = self.info()._logger_details()
         return self._logger_details
 
+    @property
+    def payload_converter(self) -> temporalio.converter.PayloadConverter:
+        if not self._payload_converter:
+            if isinstance(
+                self.payload_converter_class_or_instance,
+                temporalio.converter.PayloadConverter,
+            ):
+                self._payload_converter = self.payload_converter_class_or_instance
+            else:
+                self._payload_converter = self.payload_converter_class_or_instance()
+        return self._payload_converter
+
 
 @dataclass
 class _CompositeEvent:
@@ -339,6 +369,14 @@ class _CompleteAsyncError(BaseException):
     pass
 
 
+def payload_converter() -> temporalio.converter.PayloadConverter:
+    """Get the payload converter for the current activity.
+
+    This is often used for dynamic activities to convert payloads.
+    """
+    return _Context.current().payload_converter
+
+
 class LoggerAdapter(logging.LoggerAdapter):
     """Adapter that adds details to the log about the running activity.
 
@@ -387,9 +425,9 @@ def base_logger(self) -> logging.Logger:
 """Logger that will have contextual activity details embedded."""
 
 
-@dataclass
+@dataclass(frozen=True)
 class _Definition:
-    name: str
+    name: Optional[str]
     fn: Callable
     is_async: bool
     no_thread_cancel_exception: bool
@@ -420,7 +458,10 @@ def must_from_callable(fn: Callable) -> _Definition:
 
     @staticmethod
     def _apply_to_callable(
-        fn: Callable, *, activity_name: str, no_thread_cancel_exception: bool = False
+        fn: Callable,
+        *,
+        activity_name: Optional[str],
+        no_thread_cancel_exception: bool = False,
     ) -> None:
         # Validate the activity
         if hasattr(fn, "__temporal_activity_definition"):
@@ -447,6 +488,16 @@ def _apply_to_callable(
 
     def __post_init__(self) -> None:
         if self.arg_types is None and self.ret_type is None:
+            dynamic = self.name is None
             arg_types, ret_type = temporalio.common._type_hints_from_func(self.fn)
+            # If dynamic, must be a sequence of raw values
+            if dynamic and (
+                not arg_types
+                or len(arg_types) != 1
+                or arg_types[0] != Sequence[temporalio.common.RawValue]
+            ):
+                raise TypeError(
+                    "Dynamic activity must accept a single Sequence[temporalio.common.RawValue]"
+                )
             object.__setattr__(self, "arg_types", arg_types)
             object.__setattr__(self, "ret_type", ret_type)

temporalio/client.py

@@ -408,6 +408,8 @@ async def start_workflow(
             name = workflow
         elif callable(workflow):
             defn = temporalio.workflow._Definition.must_from_run_fn(workflow)
+            if not defn.name:
+                raise ValueError("Cannot invoke dynamic workflow explicitly")
             name = defn.name
             if result_type is None:
                 result_type = defn.ret_type
@@ -2827,6 +2829,8 @@ def __init__(
             # Use definition if callable
             if callable(workflow):
                 defn = temporalio.workflow._Definition.must_from_run_fn(workflow)
+                if not defn.name:
+                    raise ValueError("Cannot schedule dynamic workflow explicitly")
                 workflow = defn.name
             elif not isinstance(workflow, str):
                 raise TypeError("Workflow must be a string or callable")

temporalio/common.py

@@ -138,6 +138,34 @@ class QueryRejectCondition(IntEnum):
     )
 
 
+@dataclass(frozen=True)
+class RawValue:
+    """Representation of an unconverted, raw payload.
+
+    This type can be used as a parameter or return type in workflows,
+    activities, signals, and queries to pass through a raw payload.
+    Encoding/decoding of the payload is still done by the system.
+    """
+
+    payload: temporalio.api.common.v1.Payload
+
+    def __getstate__(self) -> object:
+        """Pickle support."""
+        # We'll convert payload to bytes and prepend a version number just in
+        # case we want to extend in the future
+        return b"1" + self.payload.SerializeToString()
+
+    def __setstate__(self, state: object) -> None:
+        """Pickle support."""
+        if not isinstance(state, bytes):
+            raise TypeError(f"Expected bytes state, got {type(state)}")
+        if not state[:1] == b"1":
+            raise ValueError("Bad version prefix")
+        object.__setattr__(
+            self, "payload", temporalio.api.common.v1.Payload.FromString(state[1:])
+        )
+
+
 # We choose to make this a list instead of an sequence so we can catch if people
 # are not sending lists each time but maybe accidentally sending a string (which
 # is a sequence)

temporalio/converter.py

@@ -31,6 +31,7 @@
     TypeVar,
     Union,
     get_type_hints,
+    overload,
 )
 
 import google.protobuf.json_format
@@ -43,6 +44,7 @@
 import temporalio.api.failure.v1
 import temporalio.common
 import temporalio.exceptions
+import temporalio.types
 
 if sys.version_info < (3, 11):
     # Python's datetime.fromisoformat doesn't support certain formats pre-3.11
@@ -67,6 +69,9 @@ def to_payloads(
     ) -> List[temporalio.api.common.v1.Payload]:
         """Encode values into payloads.
 
+        Implementers are expected to just return the payload for
+        :py:class:`temporalio.common.RawValue`.
+
         Args:
             values: Values to be converted.
 
@@ -88,6 +93,9 @@ def from_payloads(
     ) -> List[Any]:
         """Decode payloads into values.
 
+        Implementers are expected to treat a type hint of
+        :py:class:`temporalio.common.RawValue` as just the raw value.
+
         Args:
             payloads: Payloads to convert to Python values.
             type_hints: Types that are expected if any. This may not have any
@@ -122,6 +130,51 @@ def from_payloads_wrapper(
             return []
         return self.from_payloads(payloads.payloads)
 
+    def to_payload(self, value: Any) -> temporalio.api.common.v1.Payload:
+        """Convert a single value to a payload.
+
+        This is a shortcut for :py:meth:`to_payloads` with a single-item list
+        and result.
+
+        Args:
+            value: Value to convert to a single payload.
+
+        Returns:
+            Single converted payload.
+        """
+        return self.to_payloads([value])[0]
+
+    @overload
+    def from_payload(self, payload: temporalio.api.common.v1.Payload) -> Any:
+        ...
+
+    @overload
+    def from_payload(
+        self,
+        payload: temporalio.api.common.v1.Payload,
+        type_hint: Type[temporalio.types.AnyType],
+    ) -> temporalio.types.AnyType:
+        ...
+
+    def from_payload(
+        self,
+        payload: temporalio.api.common.v1.Payload,
+        type_hint: Optional[Type] = None,
+    ) -> Any:
+        """Convert a single payload to a value.
+
+        This is a shortcut for :py:meth:`from_payloads` with a single-item list
+        and result.
+
+        Args:
+            payload: Payload to convert to value.
+            type_hint: Optional type hint to say which type to convert to.
+
+        Returns:
+            Single converted value.
+        """
+        return self.from_payloads([payload], [type_hint] if type_hint else None)[0]
+
 
 class EncodingPayloadConverter(ABC):
     """Base converter to/from single payload/value with a known encoding for use in CompositePayloadConverter."""
@@ -209,10 +262,14 @@ def to_payloads(
             # We intentionally attempt these serially just in case a stateful
             # converter may rely on the previous values
             payload = None
-            for converter in self.converters.values():
-                payload = converter.to_payload(value)
-                if payload is not None:
-                    break
+            # RawValue should just pass through
+            if isinstance(value, temporalio.common.RawValue):
+                payload = value.payload
+            else:
+                for converter in self.converters.values():
+                    payload = converter.to_payload(value)
+                    if payload is not None:
+                        break
             if payload is None:
                 raise RuntimeError(
                     f"Value at index {index} of type {type(value)} has no known converter"
@@ -235,13 +292,17 @@ def from_payloads(
         """
         values = []
         for index, payload in enumerate(payloads):
+            type_hint = None
+            if type_hints and len(type_hints) > index:
+                type_hint = type_hints[index]
+            # Raw value should just wrap
+            if type_hint == temporalio.common.RawValue:
+                values.append(temporalio.common.RawValue(payload))
+                continue
             encoding = payload.metadata.get("encoding", b"<unknown>")
             converter = self.converters.get(encoding)
             if converter is None:
                 raise KeyError(f"Unknown payload encoding {encoding.decode()}")
-            type_hint = None
-            if type_hints is not None:
-                type_hint = type_hints[index]
             try:
                 values.append(converter.from_payload(payload, type_hint))
             except RuntimeError as err:

temporalio/testing/_activity.py

@@ -12,6 +12,7 @@
 from typing_extensions import ParamSpec
 
 import temporalio.activity
+import temporalio.converter
 import temporalio.exceptions
 import temporalio.worker._activity
 
@@ -52,12 +53,18 @@ class ActivityEnvironment:
             function.
         on_heartbeat: Function called on each heartbeat invocation by the
             activity.
+        payload_converter: Payload converter set on the activity context. This
+            must be set before :py:meth:`run`. Changes after the activity has
+            started do not take effect.
     """
 
     def __init__(self) -> None:
         """Create an ActivityEnvironment for running activity code."""
         self.info = _default_info
         self.on_heartbeat: Callable[..., None] = lambda *args: None
+        self.payload_converter = (
+            temporalio.converter.DataConverter.default.payload_converter
+        )
         self._cancelled = False
         self._worker_shutdown = False
         self._activities: Set[_Activity] = set()
@@ -139,6 +146,7 @@ def __init__(
             shield_thread_cancel_exception=None
             if not self.cancel_thread_raiser
             else self.cancel_thread_raiser.shielded,
+            payload_converter_class_or_instance=env.payload_converter,
         )
         self.task: Optional[asyncio.Task] = None