Allow jobs to be declared without a default queue.

This makes it impossible to use delay() on them.  Instead, you should
use enqueue().
This commit is contained in:
Vincent Driessen 2011-11-15 09:36:18 +01:00
parent 407f3e8b38
commit b5e0377eb0
2 changed files with 68 additions and 2 deletions

View File

@ -20,8 +20,15 @@ class DelayedResult(object):
class job(object):
def __init__(self, queue_name='default'):
self.queue = Queue(queue_name)
"""The @job decorator extends the given function with two new methods:
`delay` and `enqueue`.
"""
def __init__(self, queue_name=None):
if queue_name is not None:
self.queue = Queue(queue_name)
else:
self.queue = None
def __call__(self, f):
def enqueue(queue, *args, **kwargs):
@ -36,6 +43,8 @@ class job(object):
f.enqueue = enqueue
def delay(*args, **kwargs):
if self.queue is None:
raise ValueError('This job has no default queue set.')
return f.enqueue(self.queue, *args, **kwargs)
f.delay = delay
return f

View File

@ -4,12 +4,19 @@ from blinker import signal
from redis import Redis
from rq import conn, Queue, job
# Test data
@job('my-queue')
def testjob(name=None):
if name is None:
name = 'Stranger'
return 'Hi there, %s!' % (name,)
@job() # no queue spec'ed
def queueless_job(name=None):
if name is None:
name = 'Stranger'
return 'Hi there, %s!' % (name,)
class RQTestCase(unittest.TestCase):
def setUp(self):
@ -93,5 +100,55 @@ class TestQueue(RQTestCase):
self.assertQueueContains(q, testjob)
class TestJob(RQTestCase):
def test_job_methods(self):
"""Jobs have methods to enqueue them."""
self.assertTrue(hasattr(testjob, 'delay'))
self.assertTrue(hasattr(testjob, 'enqueue'))
self.assertTrue(hasattr(queueless_job, 'delay'))
self.assertTrue(hasattr(queueless_job, 'enqueue'))
def test_queue_empty(self):
"""Detecting empty queues."""
q = Queue('my-queue')
self.assertEquals(q.empty, True)
conn.rpush('rq:my-queue', 'some val')
self.assertEquals(q.empty, False)
def test_put_work_on_queue(self):
"""Putting work on queues using delay."""
q = Queue('my-queue')
self.assertEquals(q.empty, True)
# testjob spec holds which queue this is sent to
testjob.delay()
self.assertEquals(q.empty, False)
self.assertQueueContains(q, testjob)
def test_put_work_on_queue_fails_for_queueless_jobs(self):
"""Putting work on queues using delay fails for queueless jobs."""
self.assertRaises(ValueError, queueless_job.delay, 'Rick')
def test_put_work_on_different_queue(self):
"""Putting work on alternative queues using enqueue."""
# Override testjob spec holds which queue
q = Queue('different-queue')
self.assertEquals(q.empty, True)
testjob.enqueue(q)
self.assertEquals(q.empty, False)
self.assertQueueContains(q, testjob)
def test_put_work_on_different_queue_reverse(self):
"""Putting work on specific queues using the Queue object."""
q = Queue('alt-queue')
self.assertEquals(q.empty, True)
q.enqueue(testjob)
self.assertEquals(q.empty, False)
self.assertQueueContains(q, testjob)
if __name__ == '__main__':
unittest.main()