Fix usage of overwrite, from job to the schedule request.

Signed-off-by: Albert Callarisa <[email protected]>

Author: acroca

dapr/aio/clients/grpc/client.py

@@ -1851,13 +1851,14 @@ async def get_metadata(self) -> GetMetadataResponse:
             headers=await call.initial_metadata(),
         )
 
-    async def schedule_job_alpha1(self, job: Job) -> DaprResponse:
+    async def schedule_job_alpha1(self, job: Job, overwrite: bool = False) -> 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.
+            overwrite (bool): If true, allows this job to overwrite an existing job with the same name.
 
         Returns:
             DaprResponse: Empty response indicating successful scheduling.
@@ -1879,7 +1880,7 @@ async def schedule_job_alpha1(self, job: Job) -> DaprResponse:
 
         # Convert job to proto using the Job class private method
         job_proto = job._get_proto()
-        request = api_v1.ScheduleJobRequest(job=job_proto)
+        request = api_v1.ScheduleJobRequest(job=job_proto, overwrite=overwrite)
 
         try:
             call = self._stub.ScheduleJobAlpha1(request)

dapr/clients/grpc/_jobs.py

@@ -99,7 +99,6 @@ class Job:
             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.
     """
 
     name: str
@@ -109,7 +108,6 @@ class Job:
     ttl: Optional[str] = None
     data: Optional[GrpcAny] = None
     failure_policy: Optional[FailurePolicy] = None
-    overwrite: bool = False
 
     def _get_proto(self):
         """Convert this Job instance to a Dapr Job proto message.
@@ -123,7 +121,7 @@ def _get_proto(self):
         from google.protobuf.any_pb2 import Any as GrpcAny
 
         # Build the job proto
-        job_proto = api_v1.Job(name=self.name, overwrite=self.overwrite)
+        job_proto = api_v1.Job(name=self.name)
 
         if self.schedule:
             job_proto.schedule = self.schedule
@@ -133,7 +131,6 @@ def _get_proto(self):
             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:
@@ -184,5 +181,4 @@ def _from_proto(cls, job_proto):
             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/clients/grpc/client.py

@@ -1778,13 +1778,14 @@ def converse_alpha1(
         except RpcError as err:
             raise DaprGrpcError(err) from err
 
-    def schedule_job_alpha1(self, job: Job) -> DaprResponse:
+    def schedule_job_alpha1(self, job: Job, overwrite: bool = False) -> 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.
+            overwrite (bool): If true, allows this job to overwrite an existing job with the same name.
 
         Returns:
             DaprResponse: Empty response indicating successful scheduling.
@@ -1806,7 +1807,7 @@ def schedule_job_alpha1(self, job: Job) -> DaprResponse:
 
         # Convert job to proto using the Job class private method
         job_proto = job._get_proto()
-        request = api_v1.ScheduleJobRequest(job=job_proto)
+        request = api_v1.ScheduleJobRequest(job=job_proto, overwrite=overwrite)
 
         try:
             _, call = self.retry_policy.run_rpc(self._stub.ScheduleJobAlpha1.with_call, request)

examples/jobs/job_management.py

@@ -16,10 +16,10 @@ def main():
     with DaprClient() as client:
         # Example 0: Simple job without data (works without protobuf)
         print('0. Scheduling a simple job without data...', flush=True)
-        simple_job = Job(name='simple-job', schedule='@every 30s', overwrite=True)
+        simple_job = Job(name='simple-job', schedule='@every 30s')
 
         try:
-            client.schedule_job_alpha1(simple_job)
+            client.schedule_job_alpha1(job=simple_job, overwrite=True)
             print(f'✓ Simple job scheduled successfully', flush=True)
         except Exception as e:
             print(f'✗ Failed to schedule simple job: {e}', flush=True)
@@ -33,11 +33,10 @@ def main():
             schedule='@every 30s',
             data=job_data,
             ttl='5m',
-            overwrite=True,
         )
 
         try:
-            client.schedule_job_alpha1(recurring_job)
+            client.schedule_job_alpha1(job=recurring_job, overwrite=True)
             print(f'✓ Recurring job scheduled successfully', flush=True)
         except Exception as e:
             print(f'✗ Failed to schedule recurring job: {e}', flush=True)
@@ -68,11 +67,10 @@ def main():
             schedule='@every 45s',
             data=create_job_data('Job with drop failure policy'),
             failure_policy=DropFailurePolicy(),
-            overwrite=True,
         )
 
         try:
-            client.schedule_job_alpha1(drop_policy_job)
+            client.schedule_job_alpha1(job=drop_policy_job, overwrite=True)
             print(f'✓ Job with drop failure policy scheduled successfully', flush=True)
         except Exception as e:
             print(f'✗ Failed to schedule job with drop policy: {e}', flush=True)
@@ -83,11 +81,10 @@ def main():
             schedule='@every 60s',
             data=create_job_data('Job with constant retry policy'),
             failure_policy=ConstantFailurePolicy(max_retries=3, interval_seconds=10),
-            overwrite=True,
         )
 
         try:
-            client.schedule_job_alpha1(constant_policy_job)
+            client.schedule_job_alpha1(job=constant_policy_job, overwrite=True)
             print(f'✓ Job with constant retry policy scheduled successfully', flush=True)
         except Exception as e:
             print(f'✗ Failed to schedule job with retry policy: {e}', flush=True)

tests/clients/fake_dapr_server.py

@@ -55,6 +55,7 @@ def __init__(self, grpc_port: int = 50001, http_port: int = 8080):
         self.workflow_options: Dict[str, str] = {}
         self.metadata: Dict[str, str] = {}
         self.jobs: Dict[str, api_v1.Job] = {}
+        self.job_overwrites: Dict[str, bool] = {}
         self._next_exception = None
 
     def start(self):
@@ -550,6 +551,7 @@ def ScheduleJobAlpha1(self, request, context):
 
         # Store the job
         self.jobs[request.job.name] = request.job
+        self.job_overwrites[request.job.name] = request.overwrite
 
         return empty_pb2.Empty()
 

tests/clients/test_dapr_grpc_client.py

@@ -1253,9 +1253,31 @@ def test_schedule_job_alpha1_success(self):
         # Verify job was stored in fake server
         self.assertIn('test-job', self._fake_dapr_server.jobs)
         stored_job = self._fake_dapr_server.jobs['test-job']
+        stored_job_overwrite = self._fake_dapr_server.job_overwrites['test-job']
         self.assertEqual(stored_job.name, 'test-job')
         self.assertEqual(stored_job.schedule, '@every 1m')
-        self.assertEqual(stored_job.overwrite, False)
+        self.assertEqual(stored_job_overwrite, False)
+        # Verify data field is always set (even if empty)
+        self.assertTrue(stored_job.HasField('data'))
+
+    def test_schedule_job_alpha1_success_with_overwrite(self):
+        """Test successful job scheduling."""
+        dapr = DaprGrpcClient(f'{self.scheme}localhost:{self.grpc_port}')
+        job = Job(name='test-job', schedule='@every 1m')
+
+        # Schedule the job
+        response = dapr.schedule_job_alpha1(job=job, overwrite=True)
+
+        # Verify response type
+        self.assertIsInstance(response, DaprResponse)
+
+        # Verify job was stored in fake server
+        self.assertIn('test-job', self._fake_dapr_server.jobs)
+        stored_job = self._fake_dapr_server.jobs['test-job']
+        stored_job_overwrite = self._fake_dapr_server.job_overwrites['test-job']
+        self.assertEqual(stored_job.name, 'test-job')
+        self.assertEqual(stored_job.schedule, '@every 1m')
+        self.assertEqual(stored_job_overwrite, True)
         # Verify data field is always set (even if empty)
         self.assertTrue(stored_job.HasField('data'))
 
@@ -1280,12 +1302,12 @@ def test_schedule_job_alpha1_success_with_data(self):
         # Verify job was stored in fake server with all data
         self.assertIn('test-job-with-data', self._fake_dapr_server.jobs)
         stored_job = self._fake_dapr_server.jobs['test-job-with-data']
+        stored_job_overwrite = self._fake_dapr_server.job_overwrites['test-job-with-data']
         self.assertEqual(stored_job.name, 'test-job-with-data')
         self.assertEqual(stored_job.schedule, '@every 2m')
         self.assertEqual(stored_job.repeats, 3)
         self.assertEqual(stored_job.ttl, '10m')
-        self.assertEqual(stored_job.overwrite, False)
-
+        self.assertEqual(stored_job_overwrite, False)
         # Verify data field contains the payload
         self.assertTrue(stored_job.HasField('data'))
         self.assertEqual(
@@ -1323,7 +1345,6 @@ def test_get_job_alpha1_success(self):
         self.assertEqual(retrieved_job.schedule, '@every 1m')
         self.assertEqual(retrieved_job.repeats, 5)
         self.assertEqual(retrieved_job.ttl, '1h')
-        self.assertEqual(retrieved_job.overwrite, False)
 
     def test_get_job_alpha1_validation_error(self):
         """Test validation error in job retrieval."""

tests/clients/test_dapr_grpc_client_async.py

@@ -1184,9 +1184,33 @@ async def test_schedule_job_alpha1_success(self):
         # Verify job was stored in fake server
         self.assertIn('async-test-job', self._fake_dapr_server.jobs)
         stored_job = self._fake_dapr_server.jobs['async-test-job']
+        stored_job_overwrite = self._fake_dapr_server.job_overwrites['async-test-job']
         self.assertEqual(stored_job.name, 'async-test-job')
         self.assertEqual(stored_job.schedule, '@every 1m')
-        self.assertEqual(stored_job.overwrite, False)
+        self.assertEqual(stored_job_overwrite, False)
+        # Verify data field is always set (even if empty)
+        self.assertTrue(stored_job.HasField('data'))
+
+        await dapr.close()
+
+    async def test_schedule_job_alpha1_success_with_overwrite(self):
+        """Test successful async job scheduling."""
+        dapr = DaprGrpcClientAsync(f'{self.scheme}localhost:{self.grpc_port}')
+        job = Job(name='async-test-job', schedule='@every 1m')
+
+        # Schedule the job with overwrite
+        response = await dapr.schedule_job_alpha1(job=job, overwrite=True)
+
+        # Verify response type
+        self.assertIsInstance(response, DaprResponse)
+
+        # Verify job was stored in fake server
+        self.assertIn('async-test-job', self._fake_dapr_server.jobs)
+        stored_job = self._fake_dapr_server.jobs['async-test-job']
+        stored_job_overwrite = self._fake_dapr_server.job_overwrites['async-test-job']
+        self.assertEqual(stored_job.name, 'async-test-job')
+        self.assertEqual(stored_job.schedule, '@every 1m')
+        self.assertEqual(stored_job_overwrite, True)
         # Verify data field is always set (even if empty)
         self.assertTrue(stored_job.HasField('data'))
 
@@ -1215,11 +1239,12 @@ async def test_schedule_job_alpha1_success_with_data(self):
         # Verify job was stored in fake server with all data
         self.assertIn('async-test-job-with-data', self._fake_dapr_server.jobs)
         stored_job = self._fake_dapr_server.jobs['async-test-job-with-data']
+        stored_job_overwrite = self._fake_dapr_server.job_overwrites['async-test-job-with-data']
         self.assertEqual(stored_job.name, 'async-test-job-with-data')
         self.assertEqual(stored_job.schedule, '@every 2m')
         self.assertEqual(stored_job.repeats, 3)
         self.assertEqual(stored_job.ttl, '10m')
-        self.assertEqual(stored_job.overwrite, False)
+        self.assertEqual(stored_job_overwrite, False)
 
         # Verify data field contains the payload
         self.assertTrue(stored_job.HasField('data'))
@@ -1279,7 +1304,6 @@ async def test_get_job_alpha1_success(self):
         self.assertEqual(retrieved_job.schedule, '@every 1m')
         self.assertEqual(retrieved_job.repeats, 5)
         self.assertEqual(retrieved_job.ttl, '1h')
-        self.assertEqual(retrieved_job.overwrite, False)
 
         await dapr.close()
 
@@ -1353,11 +1377,10 @@ async def test_job_lifecycle(self):
             data=data,
             repeats=10,
             ttl='30m',
-            overwrite=True,
         )
 
         # 1. Schedule the job
-        schedule_response = await dapr.schedule_job_alpha1(job)
+        schedule_response = await dapr.schedule_job_alpha1(job=job, overwrite=True)
         self.assertIsInstance(schedule_response, DaprResponse)
 
         # 2. Get the job and verify all fields
@@ -1366,7 +1389,6 @@ async def test_job_lifecycle(self):
         self.assertEqual(retrieved_job.schedule, '@every 5m')
         self.assertEqual(retrieved_job.repeats, 10)
         self.assertEqual(retrieved_job.ttl, '30m')
-        self.assertTrue(retrieved_job.overwrite)
         self.assertEqual(retrieved_job.data.value, b'{"lifecycle": "test"}')
 
         # 3. Delete the job

tests/clients/test_jobs.py

@@ -24,7 +24,6 @@ def test_job_creation(self):
         self.assertIsNone(job.due_time)
         self.assertIsNone(job.ttl)
         self.assertIsNone(job.data)
-        self.assertEqual(job.overwrite, False)
 
         # Test job with all fields
         data = GrpcAny()
@@ -37,15 +36,13 @@ def test_job_creation(self):
             due_time='2024-01-01T00:00:00Z',
             ttl='1h',
             data=data,
-            overwrite=True,
         )
         self.assertEqual(job_full.name, 'full-job')
         self.assertEqual(job_full.schedule, '0 0 * * *')
         self.assertEqual(job_full.repeats, 5)
         self.assertEqual(job_full.due_time, '2024-01-01T00:00:00Z')
         self.assertEqual(job_full.ttl, '1h')
         self.assertEqual(job_full.data, data)
-        self.assertEqual(job_full.overwrite, True)
 
     def test_job_get_proto_full(self):
         """Test _get_proto() method with all fields."""
@@ -59,7 +56,6 @@ def test_job_get_proto_full(self):
             due_time='2024-01-01T00:00:00Z',
             ttl='1h',
             data=data,
-            overwrite=True,
         )
         job_proto = job._get_proto()
 
@@ -70,7 +66,6 @@ def test_job_get_proto_full(self):
         self.assertEqual(job_proto.repeats, 5)
         self.assertEqual(job_proto.due_time, '2024-01-01T00:00:00Z')
         self.assertEqual(job_proto.ttl, '1h')
-        self.assertTrue(job_proto.overwrite)
 
         # Verify data field
         self.assertTrue(job_proto.HasField('data'))
@@ -88,7 +83,7 @@ def test_job_get_proto_no_data(self):
     def test_job_from_proto_no_data(self):
         """Test _from_proto() method with minimal proto."""
         # Create minimal proto
-        job_proto = api_v1.Job(name='test-job', overwrite=False)
+        job_proto = api_v1.Job(name='test-job')
         job_proto.data.CopyFrom(GrpcAny())  # Empty data
 
         # Convert to Job
@@ -101,7 +96,6 @@ def test_job_from_proto_no_data(self):
         self.assertIsNone(job.due_time)
         self.assertIsNone(job.ttl)
         self.assertIsNone(job.data)  # Empty data becomes None
-        self.assertEqual(job.overwrite, False)
 
     def test_job_from_proto_full(self):
         """Test _from_proto() method with all fields."""
@@ -115,7 +109,6 @@ def test_job_from_proto_full(self):
             repeats=5,
             due_time='2024-01-01T00:00:00Z',
             ttl='1h',
-            overwrite=True,
         )
         job_proto.data.CopyFrom(data)
 
@@ -129,7 +122,6 @@ def test_job_from_proto_full(self):
         self.assertEqual(job.due_time, '2024-01-01T00:00:00Z')
         self.assertEqual(job.ttl, '1h')
         self.assertEqual(job.data.value, b'{"message": "test"}')
-        self.assertTrue(job.overwrite)
 
     def test_job_with_drop_failure_policy(self):
         """Test Job with DropFailurePolicy."""