Check that arguments are compatible with a task before scheduling

Tries to verify that the arguments passed while scheduling a job are
compatible with the task function to prevent a user from scheduling a
job that won't be able to be processed by workers.

Fixes #4

Author: NicolasLM

spinach/engine.py

@@ -85,6 +85,8 @@ def schedule_at(self, task: Schedulable, at: datetime, *args, **kwargs):
         task = self._tasks.get(task)
         job = Job(task.name, task.queue, at, task.max_retries, task_args=args,
                   task_kwargs=kwargs)
+        job.task_func = task.func
+        job.check_signature()
         return self._broker.enqueue_jobs([job])
 
     def schedule_batch(self, batch: Batch):
@@ -98,10 +100,14 @@ def schedule_batch(self, batch: Batch):
         jobs = list()
         for task, at, args, kwargs in batch.jobs_to_create:
             task = self._tasks.get(task)
-            jobs.append(
-                Job(task.name, task.queue, at, task.max_retries,
-                    task_args=args, task_kwargs=kwargs)
+            job = Job(
+                task.name, task.queue, at, task.max_retries,
+                task_args=args, task_kwargs=kwargs
             )
+            job.task_func = task.func
+            job.check_signature()
+            jobs.append(job)
+
         return self._broker.enqueue_jobs(jobs)
 
     def _arbiter_func(self, stop_when_queue_empty=False):

spinach/exc.py

@@ -4,3 +4,7 @@ class SpinachError(Exception):
 
 class UnknownTask(SpinachError):
     """Task name is not registered with the Engine."""
+
+
+class InvalidJobSignatureError(SpinachError):
+    """Job does not have proper arguments to execute the task function."""

spinach/job.py

@@ -1,12 +1,14 @@
 from datetime import datetime, timezone
 import enum
+import inspect
 import json
 from logging import getLogger
 import math
 from typing import Optional
 import uuid
 
 from . import signals
+from .exc import InvalidJobSignatureError
 from .task import RetryException, AbortException
 from .utils import human_duration, exponential_backoff
 
@@ -132,6 +134,30 @@ def deserialize(cls, job_json_string: str):
         job.retries = job_dict['retries']
         return job
 
+    def check_signature(self):
+        """Check if a job has the correct params to be executed.
+
+        This can be used to prevent the scheduling of a job that will fail
+        during execution because its arguments do not match the task function.
+
+        :raises InvalidJobSignatureError: Job arguments do not match the task
+        """
+        if self.task_func is None:
+            raise ValueError(
+                'Cannot verify signature until a task function is assigned'
+            )
+
+        try:
+            sig = inspect.signature(self.task_func)
+            sig.bind(*self.task_args, **self.task_kwargs)
+        except TypeError as e:
+            msg = 'Arguments of job not compatible with task {}: {}'.format(
+                self.task_name, e
+            )
+            raise InvalidJobSignatureError(msg)
+        except ValueError:
+            logger.info('Cannot verify job signature, assuming it is correct')
+
     def __repr__(self):
         return 'Job <{} {} {}>'.format(
             self.task_name, self.status.name, self.id

tests/test_job.py

@@ -5,6 +5,7 @@
 
 from spinach import RetryException, AbortException
 from spinach.job import Job, JobStatus, advance_job_status
+from spinach.exc import InvalidJobSignatureError
 
 from .conftest import get_now, set_now
 
@@ -122,3 +123,18 @@ def test_advance_job_status(job):
     advance_job_status('namespace', job, 1.0, AbortException('kaboom'))
     assert job.max_retries == 0
     assert job.status is JobStatus.FAILED
+
+
+def test_check_signature(job):
+    def compatible_func(a, b, foo=None):
+        pass
+
+    def incompatible_func(a, bar=None):
+        pass
+
+    job.task_func = compatible_func
+    assert job.check_signature() is None
+
+    job.task_func = incompatible_func
+    with pytest.raises(InvalidJobSignatureError):
+        job.check_signature()