Merge pull request #200 from davekch/192-check-jobstore-before-handling-event

prevent DjangoResultsStoreMixin from handling events of other jobstores

Author: jcass77

django_apscheduler/jobstores.py

@@ -36,19 +36,20 @@ def start(self, scheduler, alias):
 
         self.register_event_listeners()
 
-    @classmethod
-    def handle_submission_event(cls, event: JobSubmissionEvent):
+    def handle_submission_event(self, event: JobSubmissionEvent):
         """
         Create and return new job execution instance in the database when the job is submitted to the scheduler.
 
         :param event: JobExecutionEvent instance
         :return: DjangoJobExecution ID or None if the job execution could not be logged.
         """
+        if event.jobstore != self._alias:
+            return None
         try:
             if event.code == events.EVENT_JOB_SUBMITTED:
                 # Start logging a new job execution
                 job_execution = DjangoJobExecution.atomic_update_or_create(
-                    cls.lock,
+                    self.lock,
                     event.job_id,
                     event.scheduled_run_times[0],
                     DjangoJobExecution.SENT,
@@ -63,7 +64,7 @@ def handle_submission_event(cls, event: JobSubmissionEvent):
                 )
 
                 job_execution = DjangoJobExecution.atomic_update_or_create(
-                    cls.lock,
+                    self.lock,
                     event.job_id,
                     event.scheduled_run_times[0],
                     status,
@@ -82,14 +83,15 @@ def handle_submission_event(cls, event: JobSubmissionEvent):
 
         return job_execution.id
 
-    @classmethod
-    def handle_execution_event(cls, event: JobExecutionEvent) -> Union[int, None]:
+    def handle_execution_event(self, event: JobExecutionEvent) -> Union[int, None]:
         """
         Store "successful" job execution status in the database.
 
         :param event: JobExecutionEvent instance
         :return: DjangoJobExecution ID or None if the job execution could not be logged.
         """
+        if event.jobstore != self._alias:
+            return None
         if event.code != events.EVENT_JOB_EXECUTED:
             raise NotImplementedError(
                 f"Don't know how to handle JobExecutionEvent '{event.code}'. Expected "
@@ -98,7 +100,7 @@ def handle_execution_event(cls, event: JobExecutionEvent) -> Union[int, None]:
 
         try:
             job_execution = DjangoJobExecution.atomic_update_or_create(
-                cls.lock,
+                self.lock,
                 event.job_id,
                 event.scheduled_run_time,
                 DjangoJobExecution.SUCCESS,
@@ -111,14 +113,15 @@ def handle_execution_event(cls, event: JobExecutionEvent) -> Union[int, None]:
 
         return job_execution.id
 
-    @classmethod
-    def handle_error_event(cls, event: JobExecutionEvent) -> Union[int, None]:
+    def handle_error_event(self, event: JobExecutionEvent) -> Union[int, None]:
         """
         Store "failed" job execution status in the database.
 
         :param event: JobExecutionEvent instance
         :return: DjangoJobExecution ID or None if the job execution could not be logged.
         """
+        if event.jobstore != self._alias:
+            return None
         try:
             if event.code == events.EVENT_JOB_ERROR:
 
@@ -130,7 +133,7 @@ def handle_error_event(cls, event: JobExecutionEvent) -> Union[int, None]:
                     traceback = None
 
                 job_execution = DjangoJobExecution.atomic_update_or_create(
-                    cls.lock,
+                    self.lock,
                     event.job_id,
                     event.scheduled_run_time,
                     DjangoJobExecution.ERROR,
@@ -144,7 +147,7 @@ def handle_error_event(cls, event: JobExecutionEvent) -> Union[int, None]:
                 exception = f"Run time of job '{event.job_id}' was missed!"
 
                 job_execution = DjangoJobExecution.atomic_update_or_create(
-                    cls.lock,
+                    self.lock,
                     event.job_id,
                     event.scheduled_run_time,
                     status,

tests/test_jobstores.py

@@ -28,7 +28,7 @@ def test_start_gets_scheduler_lock(self):
     @pytest.mark.django_db
     def test_handle_submission_event_not_supported_raises_exception(self, jobstore):
         event = JobSubmissionEvent(
-            events.EVENT_ALL, "test_job", jobstore, [timezone.now()]
+            events.EVENT_ALL, "test_job", jobstore._alias, [timezone.now()]
         )
 
         with pytest.raises(NotImplementedError):
@@ -46,7 +46,7 @@ def test_handle_submission_event_creates_job_execution(
             self, event_code, jobstore, create_add_job
     ):
         job = create_add_job(jobstore, dummy_job, datetime(2016, 5, 3))
-        event = JobSubmissionEvent(event_code, job.id, jobstore, [timezone.now()])
+        event = JobSubmissionEvent(event_code, job.id, jobstore._alias, [timezone.now()])
         jobstore.handle_submission_event(event)
 
         assert DjangoJobExecution.objects.filter(job_id=event.job_id).exists()
@@ -56,7 +56,7 @@ def test_handle_submission_event_for_job_that_no_longer_exists_does_not_raise_ex
             self, jobstore
     ):
         event = JobSubmissionEvent(
-            events.EVENT_JOB_SUBMITTED, "finished_job", jobstore, [timezone.now()]
+            events.EVENT_JOB_SUBMITTED, "finished_job", jobstore._alias, [timezone.now()]
         )
         jobstore.handle_submission_event(event)
 
@@ -65,7 +65,7 @@ def test_handle_submission_event_for_job_that_no_longer_exists_does_not_raise_ex
     @pytest.mark.django_db
     def test_handle_execution_event_not_supported_raises_exception(self, jobstore):
         event = JobExecutionEvent(
-            events.EVENT_ALL, "test_job", jobstore, timezone.now()
+            events.EVENT_ALL, "test_job", jobstore._alias, timezone.now()
         )
 
         with pytest.raises(NotImplementedError):
@@ -77,7 +77,7 @@ def test_handle_execution_event_creates_job_execution(
     ):
         job = create_add_job(jobstore, dummy_job, datetime(2016, 5, 3))
         event = JobExecutionEvent(
-            events.EVENT_JOB_EXECUTED, job.id, jobstore, timezone.now()
+            events.EVENT_JOB_EXECUTED, job.id, jobstore._alias, timezone.now()
         )
         jobstore.handle_execution_event(event)
 
@@ -89,7 +89,7 @@ def test_handle_execution_event_for_job_that_no_longer_exists_does_not_raise_exc
     ):
         # Test for regression https://github.com/jcass77/django-apscheduler/issues/116
         event = JobExecutionEvent(
-            events.EVENT_JOB_EXECUTED, "finished_job", jobstore, timezone.now()
+            events.EVENT_JOB_EXECUTED, "finished_job", jobstore._alias, timezone.now()
         )
         jobstore.handle_execution_event(event)
 
@@ -98,7 +98,7 @@ def test_handle_execution_event_for_job_that_no_longer_exists_does_not_raise_exc
     @pytest.mark.django_db
     def test_handle_error_event_not_supported_raises_exception(self, jobstore):
         event = JobExecutionEvent(
-            events.EVENT_ALL, "test_job", jobstore, timezone.now()
+            events.EVENT_ALL, "test_job", jobstore._alias, timezone.now()
         )
 
         with pytest.raises(NotImplementedError):
@@ -116,7 +116,7 @@ def test_handle_error_event_creates_job_execution(
         self, jobstore, create_add_job, event_code
     ):
         job = create_add_job(jobstore, dummy_job, datetime(2016, 5, 3))
-        event = JobExecutionEvent(event_code, job.id, jobstore, timezone.now())
+        event = JobExecutionEvent(event_code, job.id, jobstore._alias, timezone.now())
         jobstore.handle_error_event(event)
 
         assert DjangoJobExecution.objects.filter(job_id=event.job_id).exists()
@@ -127,7 +127,7 @@ def test_handle_error_event_no_exception_sets_exception_text(
     ):
         job = create_add_job(jobstore, dummy_job, datetime(2016, 5, 3))
         event = JobExecutionEvent(
-            events.EVENT_JOB_ERROR, job.id, jobstore, timezone.now()
+            events.EVENT_JOB_ERROR, job.id, jobstore._alias, timezone.now()
         )
         jobstore.handle_error_event(event)
 
@@ -140,7 +140,7 @@ def test_handle_error_event_for_job_that_no_longer_exists_does_not_raise_excepti
             self, jobstore
     ):
         event = JobExecutionEvent(
-            events.EVENT_JOB_ERROR, "finished_job", jobstore, timezone.now()
+            events.EVENT_JOB_ERROR, "finished_job", jobstore._alias, timezone.now()
         )
         jobstore.handle_error_event(event)