From 06d75630e0936b08b672adb054b7e9a7a0172c1f Mon Sep 17 00:00:00 2001 From: Selwin Ong Date: Wed, 22 Aug 2012 17:21:22 +0700 Subject: [PATCH] Execute job immediately if Queue(async=False) --- rq/job.py | 3 ++- rq/queue.py | 14 +++++++++++--- tests/test_queue.py | 8 +++++++- 3 files changed, 20 insertions(+), 5 deletions(-) diff --git a/rq/job.py b/rq/job.py index b229e2f..125df36 100644 --- a/rq/job.py +++ b/rq/job.py @@ -289,7 +289,8 @@ class Job(object): def perform(self): # noqa """Invokes the job function with the job arguments. """ - return self.func(*self.args, **self.kwargs) + self._result = self.func(*self.args, **self.kwargs) + return self._result # Representation diff --git a/rq/queue.py b/rq/queue.py index b790f04..0ff5861 100644 --- a/rq/queue.py +++ b/rq/queue.py @@ -41,12 +41,14 @@ class Queue(object): name = queue_key[len(prefix):] return cls(name, connection=connection) - def __init__(self, name='default', default_timeout=None, connection=None): + def __init__(self, name='default', default_timeout=None, connection=None, + async=True): self.connection = resolve_connection(connection) prefix = self.redis_queue_namespace_prefix self.name = name self._key = '%s%s' % (prefix, name) self._default_timeout = default_timeout + self._async = async @property def key(self): @@ -158,6 +160,8 @@ class Queue(object): If the `set_meta_data` argument is `True` (default), it will update the properties `origin` and `enqueued_at`. + + If Queue is instantiated with async=False, job is executed immediately. """ if set_meta_data: job.origin = self.name @@ -168,8 +172,12 @@ class Queue(object): else: job.timeout = 180 # default - job.save() - self.push_job_id(job.id) + if self._async: + job.save() + self.push_job_id(job.id) + else: + job.perform() + job.save() return job def pop_job_id(self): diff --git a/tests/test_queue.py b/tests/test_queue.py index 115e8eb..fe50318 100644 --- a/tests/test_queue.py +++ b/tests/test_queue.py @@ -1,5 +1,5 @@ from tests import RQTestCase -from tests.fixtures import Calculator, say_hello, div_by_zero +from tests.fixtures import Calculator, div_by_zero, say_hello, some_calculation from rq import Queue, get_failed_queue from rq.job import Job from rq.exceptions import InvalidJobOperationError @@ -260,3 +260,9 @@ class TestFailedQueue(RQTestCase): self.assertEqual(job.result_ttl, 10) job_from_queue = Job.fetch(job.id, connection=self.testconn) self.assertEqual(int(job_from_queue.result_ttl), 10) + + def test_async_false(self): + """Executes a job immediately if async=False.""" + q = Queue(async=False) + job = q.enqueue(some_calculation, args=(2, 3)) + self.assertEqual(job.return_value, 6) \ No newline at end of file