Adds @task(typing=False) for disabling argument checking

celery/app/task.py

@@ -167,6 +167,11 @@ class Task(object):
     #: Name of the task.
     name = None
 
+    #: Enable argument checking.
+    #: You can set this to false if you don't want the signature to be
+    #: checked when calling the task.
+    typing = True
+
     #: Maximum number of retries before giving up.  If set to :const:`None`,
     #: it will **never** stop retrying.
     max_retries = 3
@@ -487,18 +492,26 @@ class Task(object):
             headers (Dict): Message headers to be included in the message.
 
         Returns:
-            ~@AsyncResult: Future promise.
+            ~@AsyncResult: Promise of future evaluation.
+
+        Raises:
+            TypeError: If not enough arguments are passed, or too many
+                arguments are passed.  Note that signature checks may
+                be disabled by specifying ``@task(typing=False)``.
+            kombu.exceptions.OperationalError: If a connection to the
+               transport cannot be made, or if the connection is lost.
 
         Note:
             Also supports all keyword arguments supported by
             :meth:`kombu.Producer.publish`.
         """
-        try:
-            check_arguments = self.__header__
-        except AttributeError:  # pragma: no cover
-            pass
-        else:
-            check_arguments(*(args or ()), **(kwargs or {}))
+        if self.typing:
+            try:
+                check_arguments = self.__header__
+            except AttributeError:  # pragma: no cover
+                pass
+            else:
+                check_arguments(*(args or ()), **(kwargs or {}))
 
         app = self._get_app()
         if app.conf.task_always_eager:

docs/userguide/tasks.rst

@@ -457,6 +457,9 @@ The request defines the following attributes:
         current task.  If using version one of the task protocol the chain
         tasks will be in ``request.callbacks`` instead.
 
+Example
+-------
+
 An example task accessing information in the context is:
 
 .. code-block:: python
@@ -528,6 +531,48 @@ see :setting:`worker_redirect_stdouts`).
             finally:
                 sys.stdout, sys.stderr = old_outs
 
+.. _task-argument-checking:
+
+Argument checking
+-----------------
+
+.. versionadded:: 4.0
+
+Celery will verify the arguments passed when you call the task, just
+like Python does when calling a normal function:
+
+.. code-block:: pycon
+
+    >>> @app.task
+    ... def add(x, y):
+    ...     return x + y
+
+    # Calling the task with two arguments works:
+    >>> add.delay(8, 8)
+    <AsyncResult: f59d71ca-1549-43e0-be41-4e8821a83c0c>
+
+    # Calling the task with only one argument fails:
+    >>> add.delay(8)
+    Traceback (most recent call last):
+      File "<stdin>", line 1, in <module>
+      File "celery/app/task.py", line 376, in delay
+        return self.apply_async(args, kwargs)
+      File "celery/app/task.py", line 485, in apply_async
+        check_arguments(*(args or ()), **(kwargs or {}))
+    TypeError: add() takes exactly 2 arguments (1 given)
+
+You can disable the argument checking for any task by setting its
+:attr:`~@Task.typing` attribute to :const:`False`:
+
+.. code-block:: pycon
+
+    >>> @app.task(typing=False)
+    ... def add(x, y):
+    ...     return x + y
+
+    # Works locally, but the worker reciving the task will raise an error.
+    >>> add.delay(8)
+    <AsyncResult: f59d71ca-1549-43e0-be41-4e8821a83c0c>
 
 Hiding sensitive information in arguments
 -----------------------------------------

t/unit/tasks/test_tasks.py

@@ -269,6 +269,30 @@ class test_tasks(TasksCase):
     def now(self):
         return self.app.now()
 
+    def test_typing(self):
+        @self.app.task()
+        def add(x, y, kw=1):
+            pass
+
+        with pytest.raises(TypeError):
+            add.delay(1)
+
+        with pytest.raises(TypeError):
+            add.delay(1, kw=2)
+
+        with pytest.raises(TypeError):
+            add.delay(1, 2, foobar=3)
+
+        add.delay(2, 2)
+
+    def test_typing__disabled(self):
+        @self.app.task(typing=False)
+        def add(x, y, kw=1):
+            pass
+        add.delay(1)
+        add.delay(1, kw=2)
+        add.delay(1, 2, foobar=3)
+
     @pytest.mark.usefixtures('depends_on_current_app')
     def test_unpickle_task(self):
         import pickle