PR feedback

Author: andystaples

durabletask/__init__.py

@@ -3,8 +3,22 @@
 
 """Durable Task SDK for Python"""
 
-from durabletask.worker import ConcurrencyOptions, VersioningOptions, WorkItemFilters
+from durabletask.worker import (
+    ActivityWorkItemFilter,
+    ConcurrencyOptions,
+    EntityWorkItemFilter,
+    OrchestrationWorkItemFilter,
+    VersioningOptions,
+    WorkItemFilters,
+)
 
-__all__ = ["ConcurrencyOptions", "VersioningOptions", "WorkItemFilters"]
+__all__ = [
+    "ActivityWorkItemFilter",
+    "ConcurrencyOptions",
+    "EntityWorkItemFilter",
+    "OrchestrationWorkItemFilter",
+    "VersioningOptions",
+    "WorkItemFilters",
+]
 
 PACKAGE_NAME = "durabletask"

durabletask/testing/in_memory_backend.py

@@ -594,7 +594,11 @@ def GetWorkItems(self, request: pb.GetWorkItemsRequest, context):
                                             skipped_entities.append(entity_id)
                                             continue
                                     except ValueError:
-                                        pass
+                                        self._logger.warning(
+                                            f"Cannot parse entity ID '{entity_id}' "
+                                            f"for filter matching; skipping")
+                                        skipped_entities.append(entity_id)
+                                        continue
 
                                 # Skip if this entity is already being processed
                                 if entity_id in self._entity_in_flight:

durabletask/worker.py

@@ -9,6 +9,7 @@
 import random
 import time
 from concurrent.futures import ThreadPoolExecutor
+from dataclasses import dataclass, field
 from datetime import datetime, timedelta, timezone
 from threading import Event, Thread
 from types import GeneratorType
@@ -145,49 +146,43 @@ def __init__(self, version: Optional[str] = None,
 _AUTO_GENERATE_FILTERS = object()
 
 
+@dataclass(frozen=True)
 class OrchestrationWorkItemFilter:
     """Specifies a filter for orchestration work items."""
 
-    def __init__(self, name: str, versions: Optional[list[str]] = None):
-        """Initialize an orchestration filter.
-
-        Args:
-            name: The name of the orchestration to filter.
-            versions: Optional list of versions to filter.
-        """
-        self.name = name
-        self.versions: list[str] = versions if versions is not None else []
+    name: str
+    """The name of the orchestration to filter."""
+    versions: list[str] = field(default_factory=list)
+    """Optional list of versions to filter."""
 
 
+@dataclass(frozen=True)
 class ActivityWorkItemFilter:
     """Specifies a filter for activity work items."""
 
-    def __init__(self, name: str, versions: Optional[list[str]] = None):
-        """Initialize an activity filter.
-
-        Args:
-            name: The name of the activity to filter.
-            versions: Optional list of versions to filter.
-        """
-        self.name = name
-        self.versions: list[str] = versions if versions is not None else []
+    name: str
+    """The name of the activity to filter."""
+    versions: list[str] = field(default_factory=list)
+    """Optional list of versions to filter."""
 
 
+@dataclass(frozen=True)
 class EntityWorkItemFilter:
-    """Specifies a filter for entity work items."""
+    """Specifies a filter for entity work items.
 
-    def __init__(self, name: str):
-        """Initialize an entity filter.
+    The name is normalized to lowercase to match entity registration
+    and instance ID conventions.
+    """
 
-        Args:
-            name: The name of the entity to filter.
-                  The name is normalized to lowercase to match
-                  entity registration and instance ID conventions.
-        """
-        EntityInstanceId.validate_entity_name(name)
-        self.name = name.lower()
+    name: str
+    """The name of the entity to filter."""
 
+    def __post_init__(self):
+        EntityInstanceId.validate_entity_name(self.name)
+        object.__setattr__(self, 'name', self.name.lower())
 
+
+@dataclass(frozen=True)
 class WorkItemFilters:
     """Work item filters for a Durable Task Worker.
 
@@ -199,28 +194,12 @@ class WorkItemFilters:
     :meth:`TaskHubGrpcWorker.use_work_item_filters` to enable filtering.
     """
 
-    def __init__(
-        self,
-        orchestrations: Optional[list[OrchestrationWorkItemFilter]] = None,
-        activities: Optional[list[ActivityWorkItemFilter]] = None,
-        entities: Optional[list[EntityWorkItemFilter]] = None,
-    ):
-        """Initialize work item filters.
-
-        Args:
-            orchestrations: List of orchestration filters.
-            activities: List of activity filters.
-            entities: List of entity filters.
-        """
-        self.orchestrations: list[OrchestrationWorkItemFilter] = (
-            orchestrations if orchestrations is not None else []
-        )
-        self.activities: list[ActivityWorkItemFilter] = (
-            activities if activities is not None else []
-        )
-        self.entities: list[EntityWorkItemFilter] = (
-            entities if entities is not None else []
-        )
+    orchestrations: list[OrchestrationWorkItemFilter] = field(default_factory=list)
+    """List of orchestration filters."""
+    activities: list[ActivityWorkItemFilter] = field(default_factory=list)
+    """List of activity filters."""
+    entities: list[EntityWorkItemFilter] = field(default_factory=list)
+    """List of entity filters."""
 
     @classmethod
     def _from_registry(cls, registry: '_Registry') -> 'WorkItemFilters':

examples/work_item_filtering.py

@@ -100,5 +100,3 @@ def farewell_orchestrator(ctx: task.OrchestrationContext, name: str):
         print(f"  Completed: {state.serialized_output}")
     elif state:
         print(f"  Failed: {state.failure_details}")
-
-    exit()

tests/durabletask-azuremanaged/test_dts_work_item_filters_e2e.py

@@ -13,6 +13,8 @@
     ActivityWorkItemFilter,
     EntityWorkItemFilter,
     OrchestrationWorkItemFilter,
+    VersioningOptions,
+    VersionMatchStrategy,
     WorkItemFilters,
 )
 from durabletask.azuremanaged.client import DurableTaskSchedulerClient
@@ -262,3 +264,139 @@ def ping(self, _):
 
     assert matched_invoked
     assert not unmatched_invoked
+
+
+# ------------------------------------------------------------------
+# Tests: version-aware filtering with strict versioning
+# ------------------------------------------------------------------
+
+def _simple_v2_orchestrator(ctx: task.OrchestrationContext, input: int):
+    """Orchestrator that returns immediately (no activities) for version tests."""
+    return input + 1
+
+
+def test_strict_version_matching_orchestration_completes():
+    """Orchestration scheduled with the matching version is processed."""
+    with DurableTaskSchedulerWorker(host_address=endpoint, secure_channel=True,
+                                    taskhub=taskhub_name, token_credential=None) as w:
+        w.add_orchestrator(_simple_v2_orchestrator)
+        w.use_versioning(VersioningOptions(
+            version="2.0",
+            match_strategy=VersionMatchStrategy.STRICT,
+        ))
+        w.use_work_item_filters()  # auto-generate with version
+        w.start()
+
+        c = DurableTaskSchedulerClient(host_address=endpoint, secure_channel=True,
+                                       taskhub=taskhub_name, token_credential=None)
+        id = c.schedule_new_orchestration(
+            _simple_v2_orchestrator, input=10, version="2.0")
+        state = c.wait_for_orchestration_completion(id, timeout=30)
+
+    assert state is not None
+    assert state.runtime_status == client.OrchestrationStatus.COMPLETED
+    assert state.serialized_output == "11"
+
+
+def test_strict_version_incompatible_orchestration_stays_pending():
+    """Orchestration with an incompatible version is not dispatched and stays pending."""
+    with DurableTaskSchedulerWorker(host_address=endpoint, secure_channel=True,
+                                    taskhub=taskhub_name, token_credential=None) as w:
+        w.add_orchestrator(_simple_v2_orchestrator)
+        w.use_versioning(VersioningOptions(
+            version="2.0",
+            match_strategy=VersionMatchStrategy.STRICT,
+        ))
+        w.use_work_item_filters()
+        w.start()
+
+        c = DurableTaskSchedulerClient(host_address=endpoint, secure_channel=True,
+                                       taskhub=taskhub_name, token_credential=None)
+
+        # Schedule with version "1.0" — incompatible with the worker's "2.0"
+        bad_id = c.schedule_new_orchestration(
+            _simple_v2_orchestrator, input=5, version="1.0")
+
+        # Schedule a compatible one so we can confirm the worker is active
+        good_id = c.schedule_new_orchestration(
+            _simple_v2_orchestrator, input=5, version="2.0")
+        good_state = c.wait_for_orchestration_completion(good_id, timeout=30)
+
+        assert good_state is not None
+        assert good_state.runtime_status == client.OrchestrationStatus.COMPLETED
+
+        # The incompatible orchestration must remain pending (not failed)
+        bad_state = c.get_orchestration_state(bad_id)
+        assert bad_state is not None
+        assert bad_state.runtime_status == client.OrchestrationStatus.PENDING
+
+
+def test_strict_version_no_version_orchestration_stays_pending():
+    """Orchestration scheduled without a version is not dispatched by a strict worker."""
+    with DurableTaskSchedulerWorker(host_address=endpoint, secure_channel=True,
+                                    taskhub=taskhub_name, token_credential=None) as w:
+        w.add_orchestrator(_simple_v2_orchestrator)
+        w.use_versioning(VersioningOptions(
+            version="2.0",
+            match_strategy=VersionMatchStrategy.STRICT,
+        ))
+        w.use_work_item_filters()
+        w.start()
+
+        c = DurableTaskSchedulerClient(host_address=endpoint, secure_channel=True,
+                                       taskhub=taskhub_name, token_credential=None)
+
+        # Schedule without any version
+        no_ver_id = c.schedule_new_orchestration(
+            _simple_v2_orchestrator, input=1)
+
+        # Schedule a compatible one to prove the worker is running
+        good_id = c.schedule_new_orchestration(
+            _simple_v2_orchestrator, input=1, version="2.0")
+        good_state = c.wait_for_orchestration_completion(good_id, timeout=30)
+        assert good_state is not None
+        assert good_state.runtime_status == client.OrchestrationStatus.COMPLETED
+
+        # The unversioned orchestration must remain pending
+        no_ver_state = c.get_orchestration_state(no_ver_id)
+        assert no_ver_state is not None
+        assert no_ver_state.runtime_status == client.OrchestrationStatus.PENDING
+
+
+def test_strict_version_explicit_filters_with_versions():
+    """Explicit filters with version constraints enforce strict matching."""
+    custom_filters = WorkItemFilters(
+        orchestrations=[
+            OrchestrationWorkItemFilter(
+                name=task.get_name(_simple_v2_orchestrator),
+                versions=["3.0"],
+            ),
+        ],
+    )
+
+    with DurableTaskSchedulerWorker(host_address=endpoint, secure_channel=True,
+                                    taskhub=taskhub_name, token_credential=None) as w:
+        w.add_orchestrator(_simple_v2_orchestrator)
+        w.use_work_item_filters(custom_filters)
+        w.start()
+
+        c = DurableTaskSchedulerClient(host_address=endpoint, secure_channel=True,
+                                       taskhub=taskhub_name, token_credential=None)
+
+        # Version "2.0" does not match the filter's "3.0"
+        bad_id = c.schedule_new_orchestration(
+            _simple_v2_orchestrator, input=1, version="2.0")
+
+        # Version "3.0" should match
+        good_id = c.schedule_new_orchestration(
+            _simple_v2_orchestrator, input=1, version="3.0")
+        good_state = c.wait_for_orchestration_completion(good_id, timeout=30)
+
+        assert good_state is not None
+        assert good_state.runtime_status == client.OrchestrationStatus.COMPLETED
+        assert good_state.serialized_output == "2"
+
+        # Mismatched version must remain pending
+        bad_state = c.get_orchestration_state(bad_id)
+        assert bad_state is not None
+        assert bad_state.runtime_status == client.OrchestrationStatus.PENDING