Fixes ruff

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

Author: elena-kolevska

dapr/clients/grpc/client.py

@@ -1797,6 +1797,9 @@ def schedule_job_alpha1(self, job: Job) -> DaprResponse:
         )
         validateNotBlankString(job_name=job.name)
 
+        if not job.schedule and not job.due_time:
+            raise ValueError('Job must have either schedule or due_time specified')
+
         # Convert job to proto using the Job class private method
         job_proto = job._get_proto()
         request = api_v1.ScheduleJobRequest(job=job_proto)

examples/jobs/simple_job.py

@@ -8,90 +8,85 @@
 def create_job_data(message: str):
     """Helper function to create job payload data."""
     data = GrpcAny()
-    data.value = json.dumps({"message": message}).encode('utf-8')
+    data.value = json.dumps({'message': message}).encode('utf-8')
     return data
 
 
 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
-        )
+        print('0. Scheduling a simple job without data...', flush=True)
+        simple_job = Job(name='simple-job', schedule='@every 30s', overwrite=True)
 
         try:
             client.schedule_job_alpha1(simple_job)
-            print(f"✓ Simple job scheduled successfully", flush=True)
+            print(f'✓ Simple job scheduled successfully', flush=True)
         except Exception as e:
-            print(f"✗ Failed to schedule simple job: {e}", flush=True)
+            print(f'✗ Failed to schedule simple job: {e}', flush=True)
             return
 
         # Example 1: Schedule a recurring job with cron schedule
-        print("1. Scheduling a recurring job with cron schedule...", flush=True)
-        job_data = create_job_data("Hello from recurring job!")
+        print('1. Scheduling a recurring job with cron schedule...', flush=True)
+        job_data = create_job_data('Hello from recurring job!')
         recurring_job = Job(
-            name="recurring-hello-job",
-            schedule="@every 30s",
+            name='recurring-hello-job',
+            schedule='@every 30s',
             data=job_data,
-            ttl="5m",
-            overwrite=True
+            ttl='5m',
+            overwrite=True,
         )
 
         try:
             client.schedule_job_alpha1(recurring_job)
-            print(f"✓ Recurring job scheduled successfully", flush=True)
+            print(f'✓ Recurring job scheduled successfully', flush=True)
         except Exception as e:
-            print(f"✗ Failed to schedule recurring job: {e}", flush=True)
+            print(f'✗ Failed to schedule recurring job: {e}', flush=True)
             return
 
         # Example 2: Schedule a one-time job with due_time
-        print("\n2. Scheduling a one-time job with due_time...", flush=True)
-        due_time = (datetime.now() + timedelta(seconds=10)).isoformat() + "Z"
+        print('\n2. Scheduling a one-time job with due_time...', flush=True)
+        due_time = (datetime.now() + timedelta(seconds=10)).isoformat() + 'Z'
         one_time_job = Job(
-            name="one-time-hello-job",
+            name='one-time-hello-job',
             due_time=due_time,
-            data=create_job_data("Hello from one-time job!")
+            data=create_job_data('Hello from one-time job!'),
         )
 
         try:
             client.schedule_job_alpha1(one_time_job)
-            print(f"✓ One-time job scheduled successfully", flush=True)
+            print(f'✓ One-time job scheduled successfully', flush=True)
         except Exception as e:
-            print(f"✗ Failed to schedule one-time job: {e}", flush=True)
+            print(f'✗ Failed to schedule one-time job: {e}', flush=True)
             return
 
         # Example 3: Get job details
-        print("\n3. Getting job details...", flush=True)
+        print('\n3. Getting job details...', flush=True)
         try:
-            job = client.get_job_alpha1("recurring-hello-job")
-            print(f"✓ Retrieved job details:", flush=True)
-            print(f"  - Name: {job.name}", flush=True)
-            print(f"  - Schedule: {job.schedule}", flush=True)
-            print(f"  - TTL: {job.ttl}", flush=True)
+            job = client.get_job_alpha1('recurring-hello-job')
+            print(f'✓ Retrieved job details:', flush=True)
+            print(f'  - Name: {job.name}', flush=True)
+            print(f'  - Schedule: {job.schedule}', flush=True)
+            print(f'  - TTL: {job.ttl}', flush=True)
             if job.data:
                 try:
                     payload = json.loads(job.data.value.decode('utf-8'))
-                    print(f"  - Data: {payload}", flush=True)
+                    print(f'  - Data: {payload}', flush=True)
                 except Exception:
-                    print(f"  - Data: <binary data, {len(job.data.value)} bytes>", flush=True)
+                    print(f'  - Data: <binary data, {len(job.data.value)} bytes>', flush=True)
             else:
-                print(f"  - Data: None", flush=True)
+                print(f'  - Data: None', flush=True)
         except Exception as e:
-            print(f"✗ Failed to get job details: {e}", flush=True)
+            print(f'✗ Failed to get job details: {e}', flush=True)
 
         # Example 4: Delete jobs
-        print("\n4. Cleaning up - deleting jobs...", flush=True)
-        for job_name in ["simple-job", "recurring-hello-job", "one-time-hello-job"]:
+        print('\n4. Cleaning up - deleting jobs...', flush=True)
+        for job_name in ['simple-job', 'recurring-hello-job', 'one-time-hello-job']:
             try:
                 client.delete_job_alpha1(job_name)
-                print(f"✓ Deleted job: {job_name}", flush=True)
+                print(f'✓ Deleted job: {job_name}', flush=True)
             except Exception as e:
-                print(f"✗ Failed to delete job {job_name}: {e}", flush=True)
-
+                print(f'✗ Failed to delete job {job_name}: {e}', flush=True)
 
 
-if __name__ == "__main__":
+if __name__ == '__main__':
     main()

tests/clients/fake_dapr_server.py

@@ -54,7 +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.jobs: Dict[str, api_v1.Job] = {}
         self._next_exception = None
 
     def start(self):
@@ -544,6 +544,10 @@ def ScheduleJobAlpha1(self, request, context):
         if not request.job.name:
             raise ValueError('Job name is required')
 
+        # Validate job name
+        if not request.job.schedule and not request.job.due_time:
+            raise ValueError('Schedule is empty')
+
         # Store the job
         self.jobs[request.job.name] = request.job
 

tests/clients/test_dapr_grpc_client.py

@@ -1242,7 +1242,7 @@ def test_converse_alpha1_error_handling(self):
     def test_schedule_job_alpha1_success(self):
         """Test successful job scheduling."""
         dapr = DaprGrpcClient(f'{self.scheme}localhost:{self.grpc_port}')
-        job = Job(name="test-job", schedule="@every 1m")
+        job = Job(name='test-job', schedule='@every 1m')
 
         # Schedule the job
         response = dapr.schedule_job_alpha1(job)
@@ -1251,10 +1251,10 @@ def test_schedule_job_alpha1_success(self):
         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"]
-        self.assertEqual(stored_job.name, "test-job")
-        self.assertEqual(stored_job.schedule, "@every 1m")
+        self.assertIn('test-job', self._fake_dapr_server.jobs)
+        stored_job = self._fake_dapr_server.jobs['test-job']
+        self.assertEqual(stored_job.name, 'test-job')
+        self.assertEqual(stored_job.schedule, '@every 1m')
         self.assertEqual(stored_job.overwrite, False)
         # Verify data field is always set (even if empty)
         self.assertTrue(stored_job.HasField('data'))
@@ -1269,13 +1269,7 @@ def test_schedule_job_alpha1_success_with_data(self):
         data = GrpcAny()
         data.value = b'{"message": "Hello from job!", "priority": "high"}'
 
-        job = Job(
-            name="test-job-with-data",
-            schedule="@every 2m",
-            data=data,
-            repeats=3,
-            ttl="10m"
-        )
+        job = Job(name='test-job-with-data', schedule='@every 2m', data=data, repeats=3, ttl='10m')
 
         # Schedule the job
         response = dapr.schedule_job_alpha1(job)
@@ -1284,86 +1278,88 @@ def test_schedule_job_alpha1_success_with_data(self):
         self.assertIsInstance(response, DaprResponse)
 
         # 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"]
-        self.assertEqual(stored_job.name, "test-job-with-data")
-        self.assertEqual(stored_job.schedule, "@every 2m")
+        self.assertIn('test-job-with-data', self._fake_dapr_server.jobs)
+        stored_job = self._fake_dapr_server.jobs['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.ttl, '10m')
         self.assertEqual(stored_job.overwrite, False)
 
         # Verify data field contains the payload
         self.assertTrue(stored_job.HasField('data'))
-        self.assertEqual(stored_job.data.value, b'{"message": "Hello from job!", "priority": "high"}')
+        self.assertEqual(
+            stored_job.data.value, b'{"message": "Hello from job!", "priority": "high"}'
+        )
 
     def test_schedule_job_alpha1_validation_error(self):
         """Test validation error in job scheduling."""
         dapr = DaprGrpcClient(f'{self.scheme}localhost:{self.grpc_port}')
 
         # Test empty job name - this should be caught by client validation
         with self.assertRaises(ValueError):
-            job = Job(name="", schedule="@every 1m")
+            job = Job(name='', schedule='@every 1m')
             dapr.schedule_job_alpha1(job)
 
     def test_get_job_alpha1_success(self):
         """Test successful job retrieval."""
         dapr = DaprGrpcClient(f'{self.scheme}localhost:{self.grpc_port}')
 
         # First schedule a job
-        original_job = Job(name="test-job", schedule="@every 1m", repeats=5, ttl="1h")
+        original_job = Job(name='test-job', schedule='@every 1m', repeats=5, ttl='1h')
         dapr.schedule_job_alpha1(original_job)
 
         # Now retrieve it
-        retrieved_job = dapr.get_job_alpha1("test-job")
+        retrieved_job = dapr.get_job_alpha1('test-job')
 
         # Verify response
         self.assertIsInstance(retrieved_job, Job)
-        self.assertEqual(retrieved_job.name, "test-job")
-        self.assertEqual(retrieved_job.schedule, "@every 1m")
+        self.assertEqual(retrieved_job.name, 'test-job')
+        self.assertEqual(retrieved_job.schedule, '@every 1m')
         self.assertEqual(retrieved_job.repeats, 5)
-        self.assertEqual(retrieved_job.ttl, "1h")
+        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."""
         dapr = DaprGrpcClient(f'{self.scheme}localhost:{self.grpc_port}')
 
         with self.assertRaises(ValueError):
-            dapr.get_job_alpha1("")
+            dapr.get_job_alpha1('')
 
     def test_get_job_alpha1_not_found(self):
         """Test getting a job that doesn't exist."""
         dapr = DaprGrpcClient(f'{self.scheme}localhost:{self.grpc_port}')
 
         with self.assertRaises(DaprGrpcError):
-            dapr.get_job_alpha1("non-existent-job")
+            dapr.get_job_alpha1('non-existent-job')
 
     def test_delete_job_alpha1_success(self):
         """Test successful job deletion."""
         dapr = DaprGrpcClient(f'{self.scheme}localhost:{self.grpc_port}')
 
         # First schedule a job
-        job = Job(name="test-job", schedule="@every 1m")
+        job = Job(name='test-job', schedule='@every 1m')
         dapr.schedule_job_alpha1(job)
 
         # Verify job exists
-        self.assertIn("test-job", self._fake_dapr_server.jobs)
+        self.assertIn('test-job', self._fake_dapr_server.jobs)
 
         # Delete the job
-        response = dapr.delete_job_alpha1("test-job")
+        response = dapr.delete_job_alpha1('test-job')
 
         # Verify response
         self.assertIsInstance(response, DaprResponse)
 
         # Verify job was removed from fake server
-        self.assertNotIn("test-job", self._fake_dapr_server.jobs)
+        self.assertNotIn('test-job', self._fake_dapr_server.jobs)
 
     def test_delete_job_alpha1_validation_error(self):
         """Test validation error in job deletion."""
         dapr = DaprGrpcClient(f'{self.scheme}localhost:{self.grpc_port}')
 
         with self.assertRaises(ValueError):
-            dapr.delete_job_alpha1("")
+            dapr.delete_job_alpha1('')
 
     def test_jobs_error_handling(self):
         """Test error handling for Jobs API using fake server's exception mechanism."""
@@ -1372,14 +1368,11 @@ def test_jobs_error_handling(self):
         dapr = DaprGrpcClient(f'{self.scheme}localhost:{self.grpc_port}')
 
         # Set up fake server to raise an exception on next call
-        error_status = status_pb2.Status(
-            code=code_pb2.INTERNAL,
-            message="Simulated server error"
-        )
+        error_status = status_pb2.Status(code=code_pb2.INTERNAL, message='Simulated server error')
         self._fake_dapr_server.raise_exception_on_next_call(error_status)
 
         # Try to schedule a job - should raise DaprGrpcError
-        job = Job(name="error-test", schedule="@every 1m")
+        job = Job(name='error-test', schedule='@every 1m')
         with self.assertRaises(DaprGrpcError):
             dapr.schedule_job_alpha1(job)