First commit. Sync client. Updated protos

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

Author: elena-kolevska

dapr/clients/__init__.py

@@ -19,6 +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.http.dapr_actor_http_client import DaprActorHttpClient
 from dapr.clients.http.dapr_invocation_http_client import DaprInvocationHttpClient
 from dapr.clients.retry import RetryPolicy
@@ -32,6 +33,7 @@
     'DaprActorHttpClient',
     'DaprInternalError',
     'ERROR_CODE_UNKNOWN',
+    'Job',
 ]
 
 

dapr/clients/grpc/_jobs.py

@@ -0,0 +1,110 @@
+# -*- coding: utf-8 -*-
+
+# Copyright 2025 The Dapr Authors
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#     http://www.apache.org/licenses/LICENSE-2.0
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""
+This module contains the Job class and related utilities for the Dapr Jobs API.
+"""
+
+from dataclasses import dataclass
+from typing import Optional
+
+from google.protobuf.any_pb2 import Any as GrpcAny
+
+
+@dataclass
+class Job:
+    """Represents a Dapr job for scheduling.
+
+    At least one of schedule or due_time must be provided but can also be provided together.
+
+    Attributes:
+        name (str): The unique name for the job.
+        schedule (Optional[str]): Schedule at which the job is to be run.
+            Accepts both systemd timer style cron expressions, as well as human
+            readable '@' prefixed period strings.
+        repeats (Optional[int]): The optional number of times in which the job should be
+            triggered. If not set, the job will run indefinitely or until expiration.
+        due_time (Optional[str]): The optional time at which the job should be active, or the
+            "one shot" time if other scheduling type fields are not provided. Accepts
+            a "point in time" string in the format of RFC3339, Go duration string
+            (calculated from job creation time), or non-repeating ISO8601.
+        ttl (Optional[str]): The optional time to live or expiration of the job. Accepts a
+            "point in time" string in the format of RFC3339, Go duration string
+            (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.
+        overwrite (bool): If true, allows this job to overwrite an existing job with the same name.
+    """
+
+    name: str
+    schedule: Optional[str] = None
+    repeats: Optional[int] = None
+    due_time: Optional[str] = None
+    ttl: Optional[str] = None
+    data: Optional[GrpcAny] = None
+    overwrite: bool = False
+
+    def _get_proto(self):
+        """Convert this Job instance to a Dapr Job proto message.
+
+        This is an internal method for SDK use only. Not part of the public API.
+
+        Returns:
+            api_v1.Job: The proto representation of this job.
+        """
+        from dapr.proto.runtime.v1 import dapr_pb2 as api_v1
+        from google.protobuf.any_pb2 import Any as GrpcAny
+
+        # Build the job proto
+        job_proto = api_v1.Job(name=self.name, overwrite=self.overwrite)
+
+        if self.schedule:
+            job_proto.schedule = self.schedule
+        if self.repeats is not None:
+            job_proto.repeats = self.repeats
+        if self.due_time:
+            job_proto.due_time = self.due_time
+        if self.ttl:
+            job_proto.ttl = self.ttl
+        # overwrite is already set in the constructor above
+
+        # data field is required, set empty Any if not provided
+        if self.data:
+            job_proto.data.CopyFrom(self.data)
+        else:
+            # Set empty Any proto
+            job_proto.data.CopyFrom(GrpcAny())
+
+        return job_proto
+
+    @classmethod
+    def _from_proto(cls, job_proto):
+        """Create a Job instance from a Dapr Job proto message.
+
+        This is an internal method for SDK use only. Not part of the public API.
+
+        Args:
+            job_proto (api_v1.Job): The proto message to convert.
+
+        Returns:
+            Job: A new Job instance.
+        """
+        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') else None,
+            overwrite=job_proto.overwrite,
+        )

dapr/clients/grpc/client.py

@@ -67,6 +67,7 @@
     DecryptRequestIterator,
     ConversationInput,
 )
+from dapr.clients.grpc._jobs import Job
 from dapr.clients.grpc._response import (
     BindingResponse,
     DaprResponse,
@@ -1773,6 +1774,101 @@ def converse_alpha1(
         except RpcError as err:
             raise DaprGrpcError(err) from err
 
+    def schedule_job_alpha1(self, job: Job) -> DaprResponse:
+        """Schedules a job to be triggered at a specified time or interval.
+
+        This is an Alpha API and is subject to change.
+
+        Args:
+            job (Job): The job to schedule. Must have a name and either schedule or due_time.
+
+        Returns:
+            DaprResponse: Empty response indicating successful scheduling.
+
+        Raises:
+            ValueError: If job name is empty or both schedule and due_time are missing.
+            DaprGrpcError: If the Dapr runtime returns an error.
+        """
+        # Warnings and input validation
+        warn(
+            'The Jobs API is an Alpha version and is subject to change.',
+            UserWarning,
+            stacklevel=2,
+        )
+        validateNotBlankString(job_name=job.name)
+
+        # Convert job to proto using the Job class private method
+        job_proto = job._get_proto()
+        request = api_v1.ScheduleJobRequest(job=job_proto)
+
+        try:
+            _, call = self.retry_policy.run_rpc(self._stub.ScheduleJobAlpha1.with_call, request)
+            return DaprResponse(headers=call.initial_metadata())
+        except RpcError as err:
+            raise DaprGrpcError(err) from err
+
+    def get_job_alpha1(self, name: str) -> Job:
+        """Gets a scheduled job by name.
+
+        This is an Alpha API and is subject to change.
+
+        Args:
+            name (str): The name of the job to retrieve.
+
+        Returns:
+            Job: The job details retrieved from the scheduler.
+
+        Raises:
+            ValueError: If job name is empty.
+            DaprGrpcError: If the Dapr runtime returns an error.
+        """
+        # Warnings and input validation
+        warn(
+            'The Jobs API is an Alpha version and is subject to change.',
+            UserWarning,
+            stacklevel=2,
+        )
+        validateNotBlankString(job_name=name)
+
+        request = api_v1.GetJobRequest(name=name)
+
+        try:
+            response, call = self.retry_policy.run_rpc(self._stub.GetJobAlpha1.with_call, request)
+            return Job._from_proto(response.job)
+        except RpcError as err:
+            raise DaprGrpcError(err) from err
+
+    def delete_job_alpha1(self, name: str) -> DaprResponse:
+        """Deletes a scheduled job by name.
+
+        This is an Alpha API and is subject to change.
+
+        Args:
+            name (str): The name of the job to delete.
+
+        Returns:
+            DaprResponse: Empty response indicating successful deletion.
+
+        Raises:
+            ValueError: If job name is empty.
+            DaprGrpcError: If the Dapr runtime returns an error.
+        """
+        # Warnings and input validation
+        warn(
+            'The Jobs API is an Alpha version and is subject to change.',
+            UserWarning,
+            stacklevel=2,
+        )
+        validateNotBlankString(job_name=name)
+
+        request = api_v1.DeleteJobRequest(name=name)
+
+        try:
+            _, call = self.retry_policy.run_rpc(self._stub.DeleteJobAlpha1.with_call, request)
+            return DaprResponse(headers=call.initial_metadata())
+        except RpcError as err:
+            raise DaprGrpcError(err) from err
+
     def wait(self, timeout_s: float):
         """Waits for sidecar to be available within the timeout.
 

dapr/proto/runtime/v1/dapr_pb2.py

@@ -1613,6 +1613,7 @@ class GetMetadataResponse(google.protobuf.message.Message):
     RUNTIME_VERSION_FIELD_NUMBER: builtins.int
     ENABLED_FEATURES_FIELD_NUMBER: builtins.int
     ACTOR_RUNTIME_FIELD_NUMBER: builtins.int
+    SCHEDULER_FIELD_NUMBER: builtins.int
     id: builtins.str
     runtime_version: builtins.str
     @property
@@ -1632,9 +1633,9 @@ class GetMetadataResponse(google.protobuf.message.Message):
     @property
     def enabled_features(self) -> google.protobuf.internal.containers.RepeatedScalarFieldContainer[builtins.str]: ...
     @property
-    def actor_runtime(self) -> global___ActorRuntime:
-        """TODO: Cassie: probably add scheduler runtime status"""
-
+    def actor_runtime(self) -> global___ActorRuntime: ...
+    @property
+    def scheduler(self) -> global___MetadataScheduler: ...
     def __init__(
         self,
         *,
@@ -1648,12 +1649,36 @@ class GetMetadataResponse(google.protobuf.message.Message):
         runtime_version: builtins.str = ...,
         enabled_features: collections.abc.Iterable[builtins.str] | None = ...,
         actor_runtime: global___ActorRuntime | None = ...,
+        scheduler: global___MetadataScheduler | None = ...,
     ) -> None: ...
-    def HasField(self, field_name: typing.Literal["actor_runtime", b"actor_runtime", "app_connection_properties", b"app_connection_properties"]) -> builtins.bool: ...
-    def ClearField(self, field_name: typing.Literal["active_actors_count", b"active_actors_count", "actor_runtime", b"actor_runtime", "app_connection_properties", b"app_connection_properties", "enabled_features", b"enabled_features", "extended_metadata", b"extended_metadata", "http_endpoints", b"http_endpoints", "id", b"id", "registered_components", b"registered_components", "runtime_version", b"runtime_version", "subscriptions", b"subscriptions"]) -> None: ...
+    def HasField(self, field_name: typing.Literal["_scheduler", b"_scheduler", "actor_runtime", b"actor_runtime", "app_connection_properties", b"app_connection_properties", "scheduler", b"scheduler"]) -> builtins.bool: ...
+    def ClearField(self, field_name: typing.Literal["_scheduler", b"_scheduler", "active_actors_count", b"active_actors_count", "actor_runtime", b"actor_runtime", "app_connection_properties", b"app_connection_properties", "enabled_features", b"enabled_features", "extended_metadata", b"extended_metadata", "http_endpoints", b"http_endpoints", "id", b"id", "registered_components", b"registered_components", "runtime_version", b"runtime_version", "scheduler", b"scheduler", "subscriptions", b"subscriptions"]) -> None: ...
+    def WhichOneof(self, oneof_group: typing.Literal["_scheduler", b"_scheduler"]) -> typing.Literal["scheduler"] | None: ...
 
 global___GetMetadataResponse = GetMetadataResponse
 
+@typing.final
+class MetadataScheduler(google.protobuf.message.Message):
+    """MetadataScheduler is a message that contains the list of addresses of the
+    scheduler connections.
+    """
+
+    DESCRIPTOR: google.protobuf.descriptor.Descriptor
+
+    CONNECTED_ADDRESSES_FIELD_NUMBER: builtins.int
+    @property
+    def connected_addresses(self) -> google.protobuf.internal.containers.RepeatedScalarFieldContainer[builtins.str]:
+        """connected_addresses the list of addresses of the scheduler connections."""
+
+    def __init__(
+        self,
+        *,
+        connected_addresses: collections.abc.Iterable[builtins.str] | None = ...,
+    ) -> None: ...
+    def ClearField(self, field_name: typing.Literal["connected_addresses", b"connected_addresses"]) -> None: ...
+
+global___MetadataScheduler = MetadataScheduler
+
 @typing.final
 class ActorRuntime(google.protobuf.message.Message):
     DESCRIPTOR: google.protobuf.descriptor.Descriptor
@@ -3141,6 +3166,7 @@ class Job(google.protobuf.message.Message):
     DUE_TIME_FIELD_NUMBER: builtins.int
     TTL_FIELD_NUMBER: builtins.int
     DATA_FIELD_NUMBER: builtins.int
+    OVERWRITE_FIELD_NUMBER: builtins.int
     name: builtins.str
     """The unique name for the job."""
     schedule: builtins.str
@@ -3180,6 +3206,8 @@ class Job(google.protobuf.message.Message):
     "point in time" string in the format of RFC3339, Go duration string
     (calculated from job creation time), or non-repeating ISO8601.
     """
+    overwrite: builtins.bool
+    """If true, allows this job to overwrite an existing job with the same name."""
     @property
     def data(self) -> google.protobuf.any_pb2.Any:
         """payload is the serialized job payload that will be sent to the recipient
@@ -3195,9 +3223,10 @@ class Job(google.protobuf.message.Message):
         due_time: builtins.str | None = ...,
         ttl: builtins.str | None = ...,
         data: google.protobuf.any_pb2.Any | None = ...,
+        overwrite: builtins.bool = ...,
     ) -> None: ...
     def HasField(self, field_name: typing.Literal["_due_time", b"_due_time", "_repeats", b"_repeats", "_schedule", b"_schedule", "_ttl", b"_ttl", "data", b"data", "due_time", b"due_time", "repeats", b"repeats", "schedule", b"schedule", "ttl", b"ttl"]) -> builtins.bool: ...
-    def ClearField(self, field_name: typing.Literal["_due_time", b"_due_time", "_repeats", b"_repeats", "_schedule", b"_schedule", "_ttl", b"_ttl", "data", b"data", "due_time", b"due_time", "name", b"name", "repeats", b"repeats", "schedule", b"schedule", "ttl", b"ttl"]) -> None: ...
+    def ClearField(self, field_name: typing.Literal["_due_time", b"_due_time", "_repeats", b"_repeats", "_schedule", b"_schedule", "_ttl", b"_ttl", "data", b"data", "due_time", b"due_time", "name", b"name", "overwrite", b"overwrite", "repeats", b"repeats", "schedule", b"schedule", "ttl", b"ttl"]) -> None: ...
     @typing.overload
     def WhichOneof(self, oneof_group: typing.Literal["_due_time", b"_due_time"]) -> typing.Literal["due_time"] | None: ...
     @typing.overload

dapr/proto/runtime/v1/dapr_pb2.pyi

@@ -54,6 +54,7 @@ def __init__(self, grpc_port: int = 50001, http_port: int = 8080):
         self.workflow_status = {}
         self.workflow_options: Dict[str, str] = {}
         self.metadata: Dict[str, str] = {}
+        self.jobs: Dict[str, api_v1.Job] = {} 
         self._next_exception = None
 
     def start(self):
@@ -536,6 +537,44 @@ def ConverseAlpha1(self, request, context):
 
         return api_v1.ConversationResponse(contextID=request.contextID, outputs=outputs)
 
+    def ScheduleJobAlpha1(self, request, context):
+        self.check_for_exception(context)
+
+        # Validate job name
+        if not request.job.name:
+            raise ValueError('Job name is required')
+
+        # Store the job
+        self.jobs[request.job.name] = request.job
+
+        return empty_pb2.Empty()
+
+    def GetJobAlpha1(self, request, context):
+        self.check_for_exception(context)
+
+        # Validate job name
+        if not request.name:
+            raise ValueError('Job name is required')
+
+        # Check if job exists
+        if request.name not in self.jobs:
+            raise Exception(f'Job "{request.name}" not found')
+
+        return api_v1.GetJobResponse(job=self.jobs[request.name])
+
+    def DeleteJobAlpha1(self, request, context):
+        self.check_for_exception(context)
+
+        # Validate job name
+        if not request.name:
+            raise ValueError('Job name is required')
+
+        # Check if job exists (optional - some implementations might not error)
+        if request.name in self.jobs:
+            del self.jobs[request.name]
+
+        return empty_pb2.Empty()
+
     def SetMetadata(self, request: SetMetadataRequest, context):
         self.metadata[request.key] = request.value
         return empty_pb2.Empty()