Merge pull request #18 from percipient/eager-clean-ups

Minor clean-ups to eager code path (PR #16)

CHANGELOG.rst

@@ -8,6 +8,8 @@ next
 
 * Properly set the ``current_task`` when running Batch tasks.
 * Call the success signal after a successful run of the Batch task.
+* Support running tasks eagerly via the ``Task.apply()`` method. This causes
+  the task to execute with a batch of a single item.
 
 0.2 2018-04-20
 ==============

celery_batches/__init__.py

@@ -289,16 +289,24 @@ def task_message_handler(message, body, ack, reject, callbacks, **kw):
         return task_message_handler
 
     def apply(self, args=None, kwargs=None, *_args, **_kwargs):
+        """
+        Execute this task locally as a batch of size 1, by blocking until the task returns.
+
+        Arguments:
+            args (Tuple): positional arguments passed on to the task.
+        Returns:
+            celery.result.EagerResult: pre-evaluated result.
+        """
         request = SimpleRequest(
             id=_kwargs.get("task_id", uuid()),
-            name="batch application",
+            name="batch request",
             args=args or (),
             kwargs=kwargs or {},
             delivery_info=None,
             hostname="localhost",
         )
 
-        return super().apply(([request],), {}, *_args, **_kwargs)
+        return super(Batches, self).apply(([request],), {}, *_args, **_kwargs)
 
     def flush(self, requests):
         return self.apply_buffer(requests, ([SimpleRequest.from_request(r)

t/integration/tasks.py

@@ -38,6 +38,7 @@ def add(requests):
         result += request.args[0]
 
     Results().set(result)
+    return result
 
 
 @shared_task(base=Batches, flush_every=2, flush_interval=1)

t/integration/test_batches.py

@@ -53,10 +53,21 @@ def test_always_eager():
     task_always_eager = app.conf.task_always_eager
     app.conf["task_always_eager"] = True
 
-    add.delay(1)
+    result = add.delay(1)
 
     app.conf["task_always_eager"] = task_always_eager
 
+    # An EagerResult that resolve to 1 should be returned.
+    assert result.get() == 1
+    assert Results().get() == 1
+
+
+def test_apply():
+    """The batch task runs immediately, in the same thread."""
+    result = add.apply(args=(1, ))
+
+    # An EagerResult that resolve to 1 should be returned.
+    assert result.get() == 1
     assert Results().get() == 1