Adds support for failure policies

Signed-off-by: Elena Kolevska <[email protected]>

Author: elena-kolevska

dapr/aio/clients/__init__.py

@@ -18,7 +18,7 @@
 from dapr.clients.base import DaprActorClientBase
 from dapr.clients.exceptions import DaprInternalError, ERROR_CODE_UNKNOWN
 from dapr.aio.clients.grpc.client import DaprGrpcClientAsync, MetadataTuple, InvokeMethodResponse
-from dapr.clients.grpc._jobs import Job
+from dapr.clients.grpc._jobs import Job, FailurePolicy, DropFailurePolicy, ConstantFailurePolicy
 from dapr.clients.http.dapr_actor_http_client import DaprActorHttpClient
 from dapr.clients.http.dapr_invocation_http_client import DaprInvocationHttpClient
 from dapr.conf import settings
@@ -31,6 +31,9 @@
     'DaprInternalError',
     'ERROR_CODE_UNKNOWN',
     'Job',
+    'FailurePolicy',
+    'DropFailurePolicy',
+    'ConstantFailurePolicy',
 ]
 
 from grpc.aio import (  # type: ignore

dapr/clients/__init__.py

@@ -19,7 +19,7 @@
 from dapr.clients.base import DaprActorClientBase
 from dapr.clients.exceptions import DaprInternalError, ERROR_CODE_UNKNOWN
 from dapr.clients.grpc.client import DaprGrpcClient, MetadataTuple, InvokeMethodResponse
-from dapr.clients.grpc._jobs import Job
+from dapr.clients.grpc._jobs import Job, FailurePolicy, DropFailurePolicy, ConstantFailurePolicy
 from dapr.clients.http.dapr_actor_http_client import DaprActorHttpClient
 from dapr.clients.http.dapr_invocation_http_client import DaprInvocationHttpClient
 from dapr.clients.retry import RetryPolicy
@@ -34,6 +34,9 @@
     'DaprInternalError',
     'ERROR_CODE_UNKNOWN',
     'Job',
+    'FailurePolicy',
+    'DropFailurePolicy',
+    'ConstantFailurePolicy',
 ]
 
 

dapr/clients/grpc/_jobs.py

@@ -15,10 +15,64 @@
 This module contains the Job class and related utilities for the Dapr Jobs API.
 """
 
+from abc import ABC, abstractmethod
 from dataclasses import dataclass
 from typing import Optional
 
 from google.protobuf.any_pb2 import Any as GrpcAny
+from google.protobuf.duration_pb2 import Duration as GrpcDuration
+
+
+class FailurePolicy(ABC):
+    """Abstract base class for job failure policies."""
+
+    @abstractmethod
+    def _to_proto(self):
+        """Convert this failure policy to its protobuf representation."""
+        pass
+
+
+class DropFailurePolicy(FailurePolicy):
+    """A failure policy that drops the job when it fails to trigger.
+
+    When a job fails to trigger, it will be dropped and not retried.
+    """
+
+    def _to_proto(self):
+        """Convert to protobuf JobFailurePolicy with drop policy."""
+        from dapr.proto.common.v1 import common_pb2
+
+        return common_pb2.JobFailurePolicy(drop=common_pb2.JobFailurePolicyDrop())
+
+
+class ConstantFailurePolicy(FailurePolicy):
+    """A failure policy that retries the job at constant intervals.
+
+    When a job fails to trigger, it will be retried after a constant interval,
+    up to a maximum number of retries (if specified).
+
+    Args:
+        max_retries (Optional[int]): Maximum number of retries. If None, retries indefinitely.
+        interval_seconds (Optional[int]): Interval between retries in seconds. Defaults to 30.
+    """
+
+    def __init__(self, max_retries: Optional[int] = None, interval_seconds: Optional[int] = 30):
+        self.max_retries = max_retries
+        self.interval_seconds = interval_seconds
+
+    def _to_proto(self):
+        """Convert to protobuf JobFailurePolicy with constant policy."""
+        from dapr.proto.common.v1 import common_pb2
+
+        constant_policy = common_pb2.JobFailurePolicyConstant()
+
+        if self.interval_seconds is not None:
+            constant_policy.interval.CopyFrom(GrpcDuration(seconds=self.interval_seconds))
+
+        if self.max_retries is not None:
+            constant_policy.max_retries = self.max_retries
+
+        return common_pb2.JobFailurePolicy(constant=constant_policy)
 
 
 @dataclass
@@ -43,6 +97,8 @@ class Job:
             (calculated from job creation time), or non-repeating ISO8601.
         data (Optional[GrpcAny]): The serialized job payload that will be sent to the recipient
             when the job is triggered. If not provided, an empty Any proto will be used.
+        failure_policy (Optional[FailurePolicy]): The failure policy to apply when the job fails
+            to trigger. If not provided, the default behavior is determined by the Dapr runtime.
         overwrite (bool): If true, allows this job to overwrite an existing job with the same name.
     """
 
@@ -52,6 +108,7 @@ class Job:
     due_time: Optional[str] = None
     ttl: Optional[str] = None
     data: Optional[GrpcAny] = None
+    failure_policy: Optional[FailurePolicy] = None
     overwrite: bool = False
 
     def _get_proto(self):
@@ -85,6 +142,10 @@ def _get_proto(self):
             # Set empty Any proto
             job_proto.data.CopyFrom(GrpcAny())
 
+        # Set failure policy if provided
+        if self.failure_policy:
+            job_proto.failure_policy.CopyFrom(self.failure_policy._to_proto())
+
         return job_proto
 
     @classmethod
@@ -99,12 +160,29 @@ def _from_proto(cls, job_proto):
         Returns:
             Job: A new Job instance.
         """
+        # Parse failure policy if present
+        failure_policy: Optional[FailurePolicy] = None
+        if job_proto.HasField('failure_policy'):
+            policy = job_proto.failure_policy
+            if policy.HasField('drop'):
+                failure_policy = DropFailurePolicy()
+            elif policy.HasField('constant'):
+                constant = policy.constant
+                max_retries = constant.max_retries if constant.HasField('max_retries') else None
+                interval_seconds = None
+                if constant.HasField('interval'):
+                    interval_seconds = constant.interval.seconds
+                failure_policy = ConstantFailurePolicy(
+                    max_retries=max_retries, interval_seconds=interval_seconds
+                )
+
         return cls(
             name=job_proto.name,
             schedule=job_proto.schedule if job_proto.HasField('schedule') else None,
             repeats=job_proto.repeats if job_proto.HasField('repeats') else None,
             due_time=job_proto.due_time if job_proto.HasField('due_time') else None,
             ttl=job_proto.ttl if job_proto.HasField('ttl') else None,
             data=job_proto.data if job_proto.HasField('data') and job_proto.data.value else None,
+            failure_policy=failure_policy,
             overwrite=job_proto.overwrite,
         )

dapr/proto/common/v1/common_pb2.py

@@ -18,6 +18,7 @@ import builtins
 import collections.abc
 import google.protobuf.any_pb2
 import google.protobuf.descriptor
+import google.protobuf.duration_pb2
 import google.protobuf.internal.containers
 import google.protobuf.internal.enum_type_wrapper
 import google.protobuf.message
@@ -373,3 +374,67 @@ class ConfigurationItem(google.protobuf.message.Message):
     def ClearField(self, field_name: typing.Literal["metadata", b"metadata", "value", b"value", "version", b"version"]) -> None: ...
 
 global___ConfigurationItem = ConfigurationItem
+
+@typing.final
+class JobFailurePolicy(google.protobuf.message.Message):
+    """JobFailurePolicy defines the policy to apply when a job fails to trigger."""
+
+    DESCRIPTOR: google.protobuf.descriptor.Descriptor
+
+    DROP_FIELD_NUMBER: builtins.int
+    CONSTANT_FIELD_NUMBER: builtins.int
+    @property
+    def drop(self) -> global___JobFailurePolicyDrop: ...
+    @property
+    def constant(self) -> global___JobFailurePolicyConstant: ...
+    def __init__(
+        self,
+        *,
+        drop: global___JobFailurePolicyDrop | None = ...,
+        constant: global___JobFailurePolicyConstant | None = ...,
+    ) -> None: ...
+    def HasField(self, field_name: typing.Literal["constant", b"constant", "drop", b"drop", "policy", b"policy"]) -> builtins.bool: ...
+    def ClearField(self, field_name: typing.Literal["constant", b"constant", "drop", b"drop", "policy", b"policy"]) -> None: ...
+    def WhichOneof(self, oneof_group: typing.Literal["policy", b"policy"]) -> typing.Literal["drop", "constant"] | None: ...
+
+global___JobFailurePolicy = JobFailurePolicy
+
+@typing.final
+class JobFailurePolicyDrop(google.protobuf.message.Message):
+    """JobFailurePolicyDrop is a policy which drops the job tick when the job fails to trigger."""
+
+    DESCRIPTOR: google.protobuf.descriptor.Descriptor
+
+    def __init__(
+        self,
+    ) -> None: ...
+
+global___JobFailurePolicyDrop = JobFailurePolicyDrop
+
+@typing.final
+class JobFailurePolicyConstant(google.protobuf.message.Message):
+    """JobFailurePolicyConstant is a policy which retries the job at a consistent interval when the job fails to trigger."""
+
+    DESCRIPTOR: google.protobuf.descriptor.Descriptor
+
+    INTERVAL_FIELD_NUMBER: builtins.int
+    MAX_RETRIES_FIELD_NUMBER: builtins.int
+    max_retries: builtins.int
+    """max_retries is the optional maximum number of retries to attempt before giving up.
+    If unset, the Job will be retried indefinitely.
+    """
+    @property
+    def interval(self) -> google.protobuf.duration_pb2.Duration:
+        """interval is the constant delay to wait before retrying the job."""
+
+    def __init__(
+        self,
+        *,
+        interval: google.protobuf.duration_pb2.Duration | None = ...,
+        max_retries: builtins.int | None = ...,
+    ) -> None: ...
+    def HasField(self, field_name: typing.Literal["_max_retries", b"_max_retries", "interval", b"interval", "max_retries", b"max_retries"]) -> builtins.bool: ...
+    def ClearField(self, field_name: typing.Literal["_max_retries", b"_max_retries", "interval", b"interval", "max_retries", b"max_retries"]) -> None: ...
+    def WhichOneof(self, oneof_group: typing.Literal["_max_retries", b"_max_retries"]) -> typing.Literal["max_retries"] | None: ...
+
+global___JobFailurePolicyConstant = JobFailurePolicyConstant