Merge branch 'andystaples/add-functions-support' of https://github.com/andystaples/durabletask-python into andystaples/add-functions-support

Author: andystaples

CHANGELOG.md

@@ -7,6 +7,14 @@ adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
 ## Unreleased
 
+ADDED
+
+- Added `FailureDetails.is_caused_by()` to check whether a task failure was caused by a given exception type, mirroring .NET's `TaskFailureDetails.IsCausedBy<T>()` and Java's `FailureDetails.isCausedBy()`. Passing an exception type performs a base-type-aware match (a failure caused by a subclass matches its base type) against exception classes already imported in the process; passing a string matches by qualified or unqualified name.
+
+CHANGED
+
+- **Breaking:** `FailureDetails.error_type` — and the `errorType` value sent over the wire — is now the fully-qualified type name (`module.ClassName`, e.g. `builtins.ValueError`, `durabletask.task.TaskFailedError`) instead of the bare class name, matching the .NET and Java SDKs. Code that compared `error_type` against a bare name (for example `== "ValueError"`) must be updated to the qualified name or, preferably, switched to `FailureDetails.is_caused_by()`. Because this value is persisted and crosses the orchestration boundary, failures produced by older workers may still carry a bare name; `is_caused_by()` accepts both.
+
 ## v1.8.0
 
 ADDED

durabletask-azuremanaged/CHANGELOG.md

@@ -7,6 +7,8 @@ adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
 ## Unreleased
 
+- `FailureDetails.error_type` now carries the fully-qualified type name (e.g. `durabletask.task.TaskFailedError`) instead of the bare class name, and the new `FailureDetails.is_caused_by()` helper is available (both inherited from durabletask). See the core `durabletask` changelog for details, including the breaking-change notes.
+
 ## v1.8.0
 
 - Updates base dependency to durabletask v1.8.0.

durabletask/internal/helpers.py

@@ -122,6 +122,25 @@ def new_sub_orchestration_failed_event(event_id: int, ex: Exception) -> pb.Histo
     )
 
 
+def get_qualified_name(t: type) -> str:
+    """Return the fully-qualified ``module.qualname`` name of a type.
+
+    This mirrors the fully-qualified error-type names produced by the other
+    Durable Task SDKs (.NET ``Type.ToString()`` and Java ``Class.getName()``),
+    so that :attr:`FailureDetails.error_type` values are consistent across SDKs
+    and can be matched unambiguously by :meth:`FailureDetails.is_caused_by`.
+
+    Builtin exception types keep their ``builtins.`` prefix (e.g.
+    ``builtins.ValueError``), matching how .NET keeps ``System.`` and Java keeps
+    ``java.lang.``.
+    """
+    module = getattr(t, "__module__", None)
+    qualname = getattr(t, "__qualname__", None) or getattr(t, "__name__", None) or str(t)
+    if not module:
+        return qualname
+    return f"{module}.{qualname}"
+
+
 def new_failure_details(ex: Exception, _visited: set[int] | None = None) -> pb.TaskFailureDetails:
     if _visited is None:
         _visited = set()
@@ -130,7 +149,7 @@ def new_failure_details(ex: Exception, _visited: set[int] | None = None) -> pb.T
     if len(_visited) > 10 or (inner and id(inner) in _visited) or not isinstance(inner, Exception):
         inner = None
     return pb.TaskFailureDetails(
-        errorType=type(ex).__name__,
+        errorType=get_qualified_name(type(ex)),
         errorMessage=str(ex),
         stackTrace=wrappers_pb2.StringValue(value=''.join(traceback.format_tb(ex.__traceback__))),
         innerFailure=new_failure_details(inner, _visited) if inner else None

durabletask/task.py

@@ -456,6 +456,88 @@ class FailureDetails:
     error_type: str
     stack_trace: str | None
 
+    def is_caused_by(self, error_type: str | type[BaseException]) -> bool:
+        """Return ``True`` if this failure was caused by ``error_type``.
+
+        This mirrors the .NET ``TaskFailureDetails.IsCausedBy<T>()`` and Java
+        ``FailureDetails.isCausedBy(Class)`` helpers, letting a caller introspect
+        a failure instead of comparing :attr:`error_type` strings by hand::
+
+            try:
+                yield ctx.call_activity(my_activity)
+            except TaskFailedError as e:
+                if e.details.is_caused_by(ValueError):
+                    ...
+
+        ``error_type`` may be either an exception type or a name string:
+
+        * When an exception **type** is given, the check is base-type aware: it
+          returns ``True`` if the failure's type is ``error_type`` or any of its
+          subclasses. To stay safe, this only inspects exception subclasses that
+          are already imported in the current process (it never imports a type by
+          name), so a subclass whose module has not been imported yet will not be
+          matched. Passing a non-exception type raises :class:`TypeError`.
+        * When a **string** is given, it is compared by name only (no base-type
+          awareness). The name may be fully qualified (``"module.ClassName"``) or
+          unqualified (``"ClassName"``); an unqualified name matches on the class
+          name alone and therefore cannot distinguish same-named types from
+          different modules.
+
+        Only this failure's own :attr:`error_type` is considered; chained/inner
+        causes are not traversed.
+        """
+        if isinstance(error_type, str):
+            return self._name_matches(error_type)
+        # The annotation restricts callers to exception types, but this is a
+        # public API, so validate at runtime against arbitrary objects.
+        if not (isinstance(error_type, type) and issubclass(error_type, BaseException)):  # pyright: ignore[reportUnnecessaryIsInstance]
+            raise TypeError(
+                "is_caused_by() expects an exception type or a type-name string, "
+                f"but got {error_type!r}.")
+
+        # Base-type-aware match without importing the failure's type: walk the
+        # target type and its already-loaded subclasses (visited-marked to guard
+        # against diamond inheritance) and compare each candidate's qualified
+        # name against this failure's error_type.
+        visited: set[int] = set()
+        stack: list[type] = [error_type]
+        while stack:
+            candidate = stack.pop()
+            if id(candidate) in visited:
+                continue
+            visited.add(id(candidate))
+            if self._type_matches(candidate):
+                return True
+            stack.extend(candidate.__subclasses__())
+        return False
+
+    def _type_matches(self, candidate: type) -> bool:
+        if not self.error_type:
+            return False
+        candidate_name = pbh.get_qualified_name(candidate)
+        if self.error_type == candidate_name:
+            return True
+        # Back-compat: a bare (non-qualified) stored name -- e.g. produced by an
+        # older SDK version or a non-Python SDK -- can only be compared by the
+        # unqualified class name.
+        if "." not in self.error_type:
+            return self.error_type == candidate_name.rsplit(".", 1)[-1]
+        return False
+
+    def _name_matches(self, name: str) -> bool:
+        stored = self.error_type
+        if not stored or not name:
+            return False
+        if stored == name:
+            return True
+        # When both names are fully qualified, require an exact match so that
+        # same-named types from different modules are not confused. Otherwise,
+        # honor the "qualified or unqualified" contract by comparing the
+        # unqualified (trailing) segment.
+        if "." in stored and "." in name:
+            return False
+        return stored.rsplit(".", 1)[-1] == name.rsplit(".", 1)[-1]
+
 
 class TaskFailedError(Exception):
     """Exception type for all orchestration task failures."""

durabletask/testing/in_memory_backend.py

@@ -799,7 +799,7 @@ def CompleteOrchestratorTask(self, request: pb.OrchestratorResponse, context: gr
                 if not self._is_terminal_status(instance.status):
                     instance.status = pb.ORCHESTRATION_STATUS_FAILED
                     instance.failure_details = pb.TaskFailureDetails(
-                        errorType=type(e).__name__,
+                        errorType=helpers.get_qualified_name(type(e)),
                         errorMessage=str(e),
                         isNonRetriable=True,
                     )

tests/durabletask-azuremanaged/entities/test_dts_entity_failure_handling.py

@@ -38,7 +38,7 @@ def test_orchestrator(ctx: task.OrchestrationContext, _):
     assert state.name == task.get_name(test_orchestrator)
     assert state.instance_id == id
     assert state.failure_details is not None
-    assert state.failure_details.error_type == "TaskFailedError"
+    assert state.failure_details.error_type == "durabletask.task.TaskFailedError"
     # NOTE: Because FailureDetails does not support inner_failure, we can't verify that the inner failure type is
     # EntityOperationFailedException. In the future, we should consider adding support for inner failures in
     # FailureDetails to make this more robust. This applies to all tests in this file. For now, the error message's
@@ -72,7 +72,7 @@ def test_orchestrator(ctx: task.OrchestrationContext, _):
     assert state.name == task.get_name(test_orchestrator)
     assert state.instance_id == id
     assert state.failure_details is not None
-    assert state.failure_details.error_type == "TaskFailedError"
+    assert state.failure_details.error_type == "durabletask.task.TaskFailedError"
     assert state.failure_details.message == "Operation 'fail' on entity '@failing_entity@testEntity' failed with " \
                                             "error: Something went wrong!"
     assert state.runtime_status == client.OrchestrationStatus.FAILED

tests/durabletask-azuremanaged/test_dts_async_orchestration_e2e.py

@@ -411,7 +411,7 @@ def throw_activity_with_retry(ctx: task.ActivityContext, _):
         assert state is not None
         assert state.runtime_status == client.OrchestrationStatus.FAILED
         assert state.failure_details is not None
-        assert state.failure_details.error_type == "TaskFailedError"
+        assert state.failure_details.error_type == "durabletask.task.TaskFailedError"
         assert state.failure_details.message.startswith("Sub-orchestration task #1 failed:")
         assert state.failure_details.message.endswith("Activity task #1 failed: Kah-BOOOOM!!!")
         assert state.failure_details.stack_trace is not None
@@ -449,7 +449,7 @@ def throw_activity(ctx: task.ActivityContext, _):
         assert state is not None
         assert state.runtime_status == client.OrchestrationStatus.FAILED
         assert state.failure_details is not None
-        assert state.failure_details.error_type == "TaskFailedError"
+        assert state.failure_details.error_type == "durabletask.task.TaskFailedError"
         assert state.failure_details.message.endswith("Activity task #1 failed: Kah-BOOOOM!!!")
         assert state.failure_details.stack_trace is not None
         assert throw_activity_counter == 4
@@ -536,7 +536,7 @@ def test_orchestrator(ctx: task.OrchestrationContext, _):
     assert state.name == task.get_name(test_orchestrator)
     assert state.instance_id == id
     assert state.failure_details is not None
-    assert state.failure_details.error_type == "JsonEncodeOutputException"
+    assert state.failure_details.error_type == "durabletask.internal.json_encode_output_exception.JsonEncodeOutputException"
     assert state.failure_details.message.startswith("The orchestration result could not be encoded. Object details:")
     assert state.failure_details.message.find("This is not JSON serializable") != -1
     assert state.runtime_status == client.OrchestrationStatus.FAILED

tests/durabletask-azuremanaged/test_dts_orchestration_e2e.py

@@ -538,7 +538,7 @@ def throw_activity_with_retry(ctx: task.ActivityContext, _):
         assert state is not None
         assert state.runtime_status == client.OrchestrationStatus.FAILED
         assert state.failure_details is not None
-        assert state.failure_details.error_type == "TaskFailedError"
+        assert state.failure_details.error_type == "durabletask.task.TaskFailedError"
         assert state.failure_details.message.startswith("Sub-orchestration task #1 failed:")
         assert state.failure_details.message.endswith("Activity task #1 failed: Kah-BOOOOM!!!")
         assert state.failure_details.stack_trace is not None
@@ -580,7 +580,7 @@ def throw_activity(ctx: task.ActivityContext, _):
         assert state is not None
         assert state.runtime_status == client.OrchestrationStatus.FAILED
         assert state.failure_details is not None
-        assert state.failure_details.error_type == "TaskFailedError"
+        assert state.failure_details.error_type == "durabletask.task.TaskFailedError"
         assert state.failure_details.message.endswith("Activity task #1 failed: Kah-BOOOOM!!!")
         assert state.failure_details.stack_trace is not None
         assert throw_activity_counter == 4
@@ -667,7 +667,7 @@ def test_orchestrator(ctx: task.OrchestrationContext, _):
     assert state.name == task.get_name(test_orchestrator)
     assert state.instance_id == id
     assert state.failure_details is not None
-    assert state.failure_details.error_type == "JsonEncodeOutputException"
+    assert state.failure_details.error_type == "durabletask.internal.json_encode_output_exception.JsonEncodeOutputException"
     assert state.failure_details.message.startswith("The orchestration result could not be encoded. Object details:")
     assert state.failure_details.message.find("This is not JSON serializable") != -1
     assert state.runtime_status == client.OrchestrationStatus.FAILED

tests/durabletask/entities/test_entity_failure_handling.py

@@ -50,7 +50,7 @@ def test_orchestrator(ctx: task.OrchestrationContext, _):
     assert state.name == task.get_name(test_orchestrator)
     assert state.instance_id == id
     assert state.failure_details is not None
-    assert state.failure_details.error_type == "TaskFailedError"
+    assert state.failure_details.error_type == "durabletask.task.TaskFailedError"
     assert "Something went wrong!" in state.failure_details.message
     assert state.runtime_status == client.OrchestrationStatus.FAILED
 
@@ -77,7 +77,7 @@ def test_orchestrator(ctx: task.OrchestrationContext, _):
     assert state.name == task.get_name(test_orchestrator)
     assert state.instance_id == id
     assert state.failure_details is not None
-    assert state.failure_details.error_type == "TaskFailedError"
+    assert state.failure_details.error_type == "durabletask.task.TaskFailedError"
     assert "Something went wrong!" in state.failure_details.message
     assert state.runtime_status == client.OrchestrationStatus.FAILED