Review feedback

Author: Sushisource

temporalio/common.py

@@ -1017,59 +1017,39 @@ def __post_init__(self):
 Priority.default = Priority(priority_key=None)
 
 
-class VersioningBehavior(Enum):
+class VersioningBehavior(IntEnum):
     """Specifies when a workflow might move from a worker of one Build Id to another.
 
     WARNING: Experimental API.
     """
 
-    UNSPECIFIED = 1
+    UNSPECIFIED = (
+        temporalio.api.enums.v1.VersioningBehavior.VERSIONING_BEHAVIOR_UNSPECIFIED
+    )
     """ An unspecified versioning behavior. By default, workers opting into worker versioning will
     be required to specify a behavior. See :py:class:`temporalio.worker.WorkerDeploymentOptions`."""
-    PINNED = 2
+    PINNED = temporalio.api.enums.v1.VersioningBehavior.VERSIONING_BEHAVIOR_PINNED
     """The workflow will be pinned to the current Build ID unless manually moved."""
-    AUTO_UPGRADE = 3
+    AUTO_UPGRADE = (
+        temporalio.api.enums.v1.VersioningBehavior.VERSIONING_BEHAVIOR_AUTO_UPGRADE
+    )
     """The workflow will automatically move to the latest version (default Build ID of the task
     queue) when the next task is dispatched."""
 
-    def _to_proto(self) -> temporalio.api.enums.v1.VersioningBehavior.ValueType:
-        if self == VersioningBehavior.UNSPECIFIED:
-            return temporalio.api.enums.v1.VersioningBehavior.VERSIONING_BEHAVIOR_UNSPECIFIED
-        elif self == VersioningBehavior.PINNED:
-            return temporalio.api.enums.v1.VersioningBehavior.VERSIONING_BEHAVIOR_PINNED
-        elif self == VersioningBehavior.AUTO_UPGRADE:
-            return temporalio.api.enums.v1.VersioningBehavior.VERSIONING_BEHAVIOR_AUTO_UPGRADE
-        else:
-            raise ValueError(f"Unknown VersioningBehavior: {self}")
-
 
+@dataclass(frozen=True)
 class WorkerDeploymentVersion:
     """Represents the version of a specific worker deployment.
 
     WARNING: Experimental API.
     """
 
-    _deployment_name: str
-    _build_id: str
-
-    def __init__(self, deployment_name: str, build_id: str):
-        """Build a WorkerDeploymentVersion from a deployment name and build ID."""
-        self._deployment_name = deployment_name
-        self._build_id = build_id
-
-    @property
-    def deployment_name(self) -> str:
-        """The name of the deployment."""
-        return self._deployment_name
-
-    @property
-    def build_id(self) -> str:
-        """The Build ID of this version."""
-        return self._build_id
+    deployment_name: str
+    build_id: str
 
     def to_canonical_string(self) -> str:
         """Returns the canonical string representation of the version."""
-        return f"{self._deployment_name}.{self._build_id}"
+        return f"{self.deployment_name}.{self.build_id}"
 
     @staticmethod
     def from_canonical_string(canonical: str) -> WorkerDeploymentVersion:

temporalio/worker/__init__.py

@@ -45,7 +45,7 @@
 from ._worker import (
     Worker,
     WorkerConfig,
-    WorkerDeploymentOptions,
+    WorkerDeploymentConfig,
     WorkerDeploymentVersion,
 )
 from ._workflow_instance import (
@@ -59,7 +59,7 @@
     # Primary types
     "Worker",
     "WorkerConfig",
-    "WorkerDeploymentOptions",
+    "WorkerDeploymentConfig",
     "WorkerDeploymentVersion",
     "Replayer",
     "ReplayerConfig",

temporalio/worker/_worker.py

@@ -93,7 +93,7 @@ def __init__(
         on_fatal_error: Optional[Callable[[BaseException], Awaitable[None]]] = None,
         use_worker_versioning: bool = False,
         disable_safe_workflow_eviction: bool = False,
-        deployment_options: Optional[WorkerDeploymentOptions] = None,
+        deployment_config: Optional[WorkerDeploymentConfig] = None,
     ) -> None:
         """Create a worker to process workflows and/or activities.
 
@@ -136,8 +136,8 @@ def __init__(
             build_id: Unique identifier for the current runtime. This is best
                 set as a hash of all code and should change only when code does.
                 If unset, a best-effort identifier is generated.
-                Exclusive with `deployment_options`.
-                WARNING: Deprecated. Use `deployment_options` instead.
+                Exclusive with `deployment_config`.
+                WARNING: Deprecated. Use `deployment_config` instead.
             identity: Identity for this worker client. If unset, the client
                 identity is used.
             max_cached_workflows: If nonzero, workflows will be cached and
@@ -219,27 +219,28 @@ def __init__(
                 workflows which it claims to be compatible with. For more
                 information, see
                 https://docs.temporal.io/workers#worker-versioning.
-                Exclusive with `deployment_options`.
-                WARNING: Deprecated. Use `deployment_options` instead.
+                Exclusive with `deployment_config`.
+                WARNING: Deprecated. Use `deployment_config` instead.
             disable_safe_workflow_eviction: If true, instead of letting the
                 workflow collect its tasks properly, the worker will simply let
                 the Python garbage collector collect the tasks. WARNING: Users
                 should not set this value to true. The garbage collector will
                 throw ``GeneratorExit`` in coroutines causing them to wake up
                 in different threads and run ``finally`` and other code in the
                 wrong workflow environment.
-            deployment_options: Deployment options for the worker. Exclusive with `build_id` and
+            deployment_config: Deployment config for the worker. Exclusive with `build_id` and
                 `use_worker_versioning`.
+                WARNING: This is an experimental feature and may change in the future.
         """
         if not activities and not workflows:
             raise ValueError("At least one activity or workflow must be specified")
         if use_worker_versioning and not build_id:
             raise ValueError(
                 "build_id must be specified when use_worker_versioning is True"
             )
-        if deployment_options and (build_id or use_worker_versioning):
+        if deployment_config and (build_id or use_worker_versioning):
             raise ValueError(
-                "deployment_options cannot be used with build_id or use_worker_versioning"
+                "deployment_config cannot be used with build_id or use_worker_versioning"
             )
 
         # Prepend applicable client interceptors to the given ones
@@ -330,9 +331,9 @@ def __init__(
         self._workflow_worker: Optional[_WorkflowWorker] = None
         if workflows:
             should_enforce_versioning_behavior = (
-                deployment_options is not None
-                and deployment_options.use_worker_versioning
-                and deployment_options.default_versioning_behavior
+                deployment_config is not None
+                and deployment_config.use_worker_versioning
+                and deployment_config.default_versioning_behavior
                 == temporalio.common.VersioningBehavior.UNSPECIFIED
             )
             self._workflow_worker = _WorkflowWorker(
@@ -375,9 +376,9 @@ def __init__(
         bridge_tuner = tuner._to_bridge_tuner()
 
         versioning_strategy: temporalio.bridge.worker.WorkerVersioningStrategy
-        if deployment_options:
+        if deployment_config:
             versioning_strategy = (
-                deployment_options._to_bridge_worker_deployment_options()
+                deployment_config._to_bridge_worker_deployment_options()
             )
         elif use_worker_versioning:
             build_id = build_id or load_default_build_id()
@@ -712,12 +713,15 @@ class WorkerConfig(TypedDict, total=False):
     on_fatal_error: Optional[Callable[[BaseException], Awaitable[None]]]
     use_worker_versioning: bool
     disable_safe_workflow_eviction: bool
-    deployment_options: Optional[WorkerDeploymentOptions]
+    deployment_config: Optional[WorkerDeploymentConfig]
 
 
 @dataclass
-class WorkerDeploymentOptions:
-    """Options for configuring the Worker Versioning feature."""
+class WorkerDeploymentConfig:
+    """Options for configuring the Worker Versioning feature.
+
+    WARNING: This is an experimental feature and may change in the future.
+    """
 
     version: WorkerDeploymentVersion
     use_worker_versioning: bool
@@ -732,7 +736,7 @@ def _to_bridge_worker_deployment_options(
                 build_id=self.version.build_id,
             ),
             use_worker_versioning=self.use_worker_versioning,
-            default_versioning_behavior=self.default_versioning_behavior._to_proto(),
+            default_versioning_behavior=self.default_versioning_behavior.value,
         )
 
 

temporalio/worker/_workflow.py

@@ -144,14 +144,14 @@ def __init__(
                     if not_in_annotation:
                         raise ValueError(
                             f"Workflow {defn.name} must specify a versioning behavior using "
-                            "the `versioning_behavior` argument to `@workflow.run`."
+                            "the `versioning_behavior` argument to `@workflow.defn`."
                         )
                 else:
-                    if defn.dynamic_versioning_behavior is None:
+                    if not_in_annotation and defn.dynamic_versioning_behavior is None:
                         raise ValueError(
-                            f"Dynamic Workflow {defn.name} must specify a versioning behavior "
-                            "using `@workflow.dynamic_versioning_behavior` or the "
-                            "`versioning_behavior` argument to `@workflow.run`."
+                            f"Dynamic Workflow {defn.cls.__qualname__} must specify a versioning "
+                            "behavior using `@workflow.dynamic_versioning_behavior` or the "
+                            "`versioning_behavior` argument to `@workflow.defn`."
                         )
 
             # Prepare the workflow with the runner (this will error in the

temporalio/worker/_workflow_instance.py

@@ -346,7 +346,7 @@ def activate(
         )
         self._current_completion.successful.SetInParent()
         self._current_completion.successful.versioning_behavior = (
-            self._defn.versioning_behavior._to_proto()
+            self._defn.versioning_behavior.value
             if self._defn.versioning_behavior
             else temporalio.api.enums.v1.VersioningBehavior.VERSIONING_BEHAVIOR_UNSPECIFIED
         )
@@ -425,7 +425,7 @@ def activate(
                         != temporalio.api.enums.v1.VersioningBehavior.VERSIONING_BEHAVIOR_UNSPECIFIED
                     ):
                         self._current_completion.successful.versioning_behavior = (
-                            vb._to_proto()
+                            vb.value
                         )
 
         # If we're deleting, there better be no more tasks. It is important for

temporalio/workflow.py

@@ -90,12 +90,16 @@ def defn(
     name: Optional[str] = None,
     sandboxed: bool = True,
     failure_exception_types: Sequence[Type[BaseException]] = [],
+    versioning_behavior: temporalio.common.VersioningBehavior = temporalio.common.VersioningBehavior.UNSPECIFIED,
 ) -> Callable[[ClassType], ClassType]: ...
 
 
 @overload
 def defn(
-    *, sandboxed: bool = True, dynamic: bool = False
+    *,
+    sandboxed: bool = True,
+    dynamic: bool = False,
+    versioning_behavior: temporalio.common.VersioningBehavior = temporalio.common.VersioningBehavior.UNSPECIFIED,
 ) -> Callable[[ClassType], ClassType]: ...
 
 
@@ -106,6 +110,7 @@ def defn(
     sandboxed: bool = True,
     dynamic: bool = False,
     failure_exception_types: Sequence[Type[BaseException]] = [],
+    versioning_behavior: temporalio.common.VersioningBehavior = temporalio.common.VersioningBehavior.UNSPECIFIED,
 ):
     """Decorator for workflow classes.
 
@@ -127,6 +132,8 @@ def defn(
             applied in addition to ones set on the worker constructor. If
             ``Exception`` is set, it effectively will fail a workflow/update in
             all user exception cases. WARNING: This setting is experimental.
+        versioning_behavior: Specifies the versioning behavior to use for this workflow.
+            WARNING: This setting is experimental.
     """
 
     def decorator(cls: ClassType) -> ClassType:
@@ -136,6 +143,7 @@ def decorator(cls: ClassType) -> ClassType:
             workflow_name=name or cls.__name__ if not dynamic else None,
             sandboxed=sandboxed,
             failure_exception_types=failure_exception_types,
+            versioning_behavior=versioning_behavior,
         )
         return cls
 
@@ -164,29 +172,7 @@ def init(
     return init_fn
 
 
-@dataclass(frozen=True)
-class _RunAttributes:
-    versioning_behavior: temporalio.common.VersioningBehavior
-
-
-@overload
-def run(
-    fn: CallableAsyncType,
-) -> CallableAsyncType: ...
-
-
-@overload
-def run(
-    *,
-    versioning_behavior: temporalio.common.VersioningBehavior = temporalio.common.VersioningBehavior.UNSPECIFIED,
-) -> Callable[[CallableAsyncType], CallableAsyncType]: ...
-
-
-def run(
-    fn: Optional[CallableAsyncType] = None,
-    *,
-    versioning_behavior: temporalio.common.VersioningBehavior = temporalio.common.VersioningBehavior.UNSPECIFIED,
-):
+def run(fn: CallableAsyncType) -> CallableAsyncType:
     """Decorator for the workflow run method.
 
     This must be used on one and only one async method defined on the same class
@@ -199,36 +185,18 @@ def run(
 
     Args:
         fn: The function to decorate.
-        versioning_behavior: Specifies the versioning behavior to use for this workflow.
-            WARNING: This setting is experimental.
     """
-
-    def decorator(
-        versioning_behavior: temporalio.common.VersioningBehavior, fn: CallableAsyncType
-    ) -> CallableAsyncType:
-        if not inspect.iscoroutinefunction(fn):
-            raise ValueError("Workflow run method must be an async function")
-        # Disallow local classes because we need to have the class globally
-        # referenceable by name
-        if "<locals>" in fn.__qualname__:
-            raise ValueError(
-                "Local classes unsupported, @workflow.run cannot be on a local class"
-            )
-        setattr(
-            fn,
-            "__temporal_workflow_run",
-            _RunAttributes(versioning_behavior=versioning_behavior),
-        )
-        # TODO(cretz): Why is MyPy unhappy with this return?
-        return fn  # type: ignore[return-value]
-
-    if fn is None:
-        return partial(
-            decorator,
-            versioning_behavior,
+    if not inspect.iscoroutinefunction(fn):
+        raise ValueError("Workflow run method must be an async function")
+    # Disallow local classes because we need to have the class globally
+    # referenceable by name
+    if "<locals>" in fn.__qualname__:
+        raise ValueError(
+            "Local classes unsupported, @workflow.run cannot be on a local class"
         )
-    else:
-        return decorator(versioning_behavior, fn)
+    setattr(fn, "__temporal_workflow_run", True)
+    # TODO(cretz): Why is MyPy unhappy with this return?
+    return fn  # type: ignore[return-value]
 
 
 class HandlerUnfinishedPolicy(Enum):
@@ -1534,6 +1502,7 @@ def _apply_to_class(
         workflow_name: Optional[str],
         sandboxed: bool,
         failure_exception_types: Sequence[Type[BaseException]],
+        versioning_behavior: temporalio.common.VersioningBehavior,
     ) -> None:
         # Check it's not being doubly applied
         if _Definition.from_class(cls):
@@ -1543,14 +1512,14 @@ def _apply_to_class(
         # Collect run fn and all signal/query/update fns
         init_fn: Optional[Callable[..., None]] = None
         run_fn: Optional[Callable[..., Awaitable[Any]]] = None
-        seen_run_attr: Optional[_RunAttributes] = None
+        seen_run_attr = False
         signals: Dict[Optional[str], _SignalDefinition] = {}
         queries: Dict[Optional[str], _QueryDefinition] = {}
         updates: Dict[Optional[str], _UpdateDefinition] = {}
         dynamic_versioning_behavior: Optional[CallableSyncNoParam] = None
         for name, member in inspect.getmembers(cls):
             if hasattr(member, "__temporal_workflow_run"):
-                seen_run_attr = getattr(member, "__temporal_workflow_run")
+                seen_run_attr = True
                 if not _is_unbound_method_on_cls(member, cls):
                     issues.append(
                         f"@workflow.run method {name} must be defined on {cls.__qualname__}"
@@ -1627,8 +1596,7 @@ def _apply_to_class(
                 ):
                     continue
                 if hasattr(base_member, "__temporal_workflow_run"):
-                    # TODO: Not sure this needs to exist?
-                    # seen_run_attr = True
+                    seen_run_attr = True
                     if not run_fn or base_member.__name__ != run_fn.__name__:
                         issues.append(
                             f"@workflow.run defined on {base_member.__qualname__} but not on the override"
@@ -1683,7 +1651,7 @@ def _apply_to_class(
             updates=updates,
             sandboxed=sandboxed,
             failure_exception_types=failure_exception_types,
-            versioning_behavior=seen_run_attr.versioning_behavior,
+            versioning_behavior=versioning_behavior,
             dynamic_versioning_behavior=dynamic_versioning_behavior,
         )
         setattr(cls, "__temporal_workflow_definition", defn)