util
/
rq
mirror of https://github.com/peter4431/rq.git
4
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Deprecate job.status.

Browse Source
main
Selwin Ong 11 years ago
parent 97f8ac8c76
commit f5c3c9a6dd
5 changed files with 32 additions and 20 deletions
Show Stats Download Patch File Download Diff File

24
rq/job.py
Unescape Escape View File

@ -112,31 +112,43 @@ class Job(object):
job._dependency_id = depends_on.id if isinstance(depends_on, Job) else depends_on job._dependency_id = depends_on.id if isinstance(depends_on, Job) else depends_on
return job return job
def _get_status(self): def get_status(self):
self._status = as_text(self.connection.hget(self.key, 'status')) self._status = as_text(self.connection.hget(self.key, 'status'))
return self._status return self._status
def _set_status(self, status): def _get_status(self):
raise DeprecationWarning(
"job.status is deprecated. Use job.get_status() instead"
)
return self.get_status()
def set_status(self, status):
self._status = status self._status = status
self.connection.hset(self.key, 'status', self._status) self.connection.hset(self.key, 'status', self._status)
def _set_status(self, status):
raise DeprecationWarning(
"job.status is deprecated. Use job.set_status() instead"
)
self.set_status(status)
status = property(_get_status, _set_status) status = property(_get_status, _set_status)
@property @property
def is_finished(self): def is_finished(self):
return self.status == Status.FINISHED return self.get_status() == Status.FINISHED
@property @property
def is_queued(self): def is_queued(self):
return self.status == Status.QUEUED return self.get_status() == Status.QUEUED
@property @property
def is_failed(self): def is_failed(self):
return self.status == Status.FAILED return self.get_status() == Status.FAILED
@property @property
def is_started(self): def is_started(self):
return self.status == Status.STARTED return self.get_status() == Status.STARTED
@property @property
def dependency(self): def dependency(self):

4
rq/queue.py
Unescape Escape View File

@ -160,7 +160,7 @@ class Queue(object):
while True: while True:
try: try:
pipe.watch(depends_on.key) pipe.watch(depends_on.key)
if depends_on.status != Status.FINISHED: if depends_on.get_status() != Status.FINISHED:
job.register_dependency() job.register_dependency()
job.save() job.save()
return job return job
@ -383,7 +383,7 @@ class FailedQueue(Queue):
if self.remove(job) == 0: if self.remove(job) == 0:
raise InvalidJobOperationError('Cannot requeue non-failed jobs.') raise InvalidJobOperationError('Cannot requeue non-failed jobs.')
job.status = Status.QUEUED job.set_status(Status.QUEUED)
job.exc_info = None job.exc_info = None
q = Queue(job.origin, connection=self.connection) q = Queue(job.origin, connection=self.connection)
q.enqueue_job(job) q.enqueue_job(job)

6
rq/worker.py
Unescape Escape View File

@ -352,7 +352,7 @@ class Worker(object):
self.set_current_job_id(job.id) self.set_current_job_id(job.id)
# Use the public setter here, to immediately update Redis # Use the public setter here, to immediately update Redis
job.status = Status.STARTED job.set_status(Status.STARTED)
self.log.info('%s: %s (%s)' % (green(queue.name), self.log.info('%s: %s (%s)' % (green(queue.name),
blue(job.description), job.id)) blue(job.description), job.id))
@ -361,7 +361,7 @@ class Worker(object):
self.heartbeat() self.heartbeat()
self.set_current_job_id(None) self.set_current_job_id(None)
if job.status == Status.FINISHED: if job.get_status() == Status.FINISHED:
queue.enqueue_dependents(job) queue.enqueue_dependents(job)
did_perform_work = True did_perform_work = True
@ -476,7 +476,7 @@ class Worker(object):
except Exception: except Exception:
# Use the public setter here, to immediately update Redis # Use the public setter here, to immediately update Redis
job.status = Status.FAILED job.set_status(Status.FAILED)
self.handle_exception(job, *sys.exc_info()) self.handle_exception(job, *sys.exc_info())
return False return False

8
tests/test_queue.py
Unescape Escape View File

@ -258,7 +258,7 @@ class TestQueue(RQTestCase):
"""Enqueueing a job sets its status to "queued".""" """Enqueueing a job sets its status to "queued"."""
q = Queue() q = Queue()
job = q.enqueue(say_hello) job = q.enqueue(say_hello)
self.assertEqual(job.status, Status.QUEUED) self.assertEqual(job.get_status(), Status.QUEUED)
def test_all_queues(self): def test_all_queues(self):
"""All queues""" """All queues"""
@ -316,7 +316,7 @@ class TestQueue(RQTestCase):
self.assertEqual(q.job_ids, []) self.assertEqual(q.job_ids, [])
# Jobs dependent on finished jobs are immediately enqueued # Jobs dependent on finished jobs are immediately enqueued
parent_job.status = Status.FINISHED parent_job.set_status(Status.FINISHED)
parent_job.save() parent_job.save()
job = q.enqueue_call(say_hello, depends_on=parent_job) job = q.enqueue_call(say_hello, depends_on=parent_job)
self.assertEqual(q.job_ids, [job.id]) self.assertEqual(q.job_ids, [job.id])
@ -332,7 +332,7 @@ class TestQueue(RQTestCase):
self.assertEqual(job.timeout, 123) self.assertEqual(job.timeout, 123)
# Jobs dependent on finished jobs are immediately enqueued # Jobs dependent on finished jobs are immediately enqueued
parent_job.status = Status.FINISHED parent_job.set_status(Status.FINISHED)
parent_job.save() parent_job.save()
job = q.enqueue_call(say_hello, depends_on=parent_job, timeout=123) job = q.enqueue_call(say_hello, depends_on=parent_job, timeout=123)
self.assertEqual(q.job_ids, [job.id]) self.assertEqual(q.job_ids, [job.id])
@ -394,7 +394,7 @@ class TestFailedQueue(RQTestCase):
get_failed_queue().requeue(job.id) get_failed_queue().requeue(job.id)
job = Job.fetch(job.id) job = Job.fetch(job.id)
self.assertEqual(job.status, Status.QUEUED) self.assertEqual(job.get_status(), Status.QUEUED)
def test_enqueue_preserves_result_ttl(self): def test_enqueue_preserves_result_ttl(self):
"""Enqueueing persists result_ttl.""" """Enqueueing persists result_ttl."""

10
tests/test_worker.py
Unescape Escape View File

@ -214,14 +214,14 @@ class TestWorker(RQTestCase):
w = Worker([q]) w = Worker([q])
job = q.enqueue(say_hello) job = q.enqueue(say_hello)
self.assertEqual(job.status, Status.QUEUED) self.assertEqual(job.get_status(), Status.QUEUED)
self.assertEqual(job.is_queued, True) self.assertEqual(job.is_queued, True)
self.assertEqual(job.is_finished, False) self.assertEqual(job.is_finished, False)
self.assertEqual(job.is_failed, False) self.assertEqual(job.is_failed, False)
w.work(burst=True) w.work(burst=True)
job = Job.fetch(job.id) job = Job.fetch(job.id)
self.assertEqual(job.status, Status.FINISHED) self.assertEqual(job.get_status(), Status.FINISHED)
self.assertEqual(job.is_queued, False) self.assertEqual(job.is_queued, False)
self.assertEqual(job.is_finished, True) self.assertEqual(job.is_finished, True)
self.assertEqual(job.is_failed, False) self.assertEqual(job.is_failed, False)
@ -230,7 +230,7 @@ class TestWorker(RQTestCase):
job = q.enqueue(div_by_zero, args=(1,)) job = q.enqueue(div_by_zero, args=(1,))
w.work(burst=True) w.work(burst=True)
job = Job.fetch(job.id) job = Job.fetch(job.id)
self.assertEqual(job.status, Status.FAILED) self.assertEqual(job.get_status(), Status.FAILED)
self.assertEqual(job.is_queued, False) self.assertEqual(job.is_queued, False)
self.assertEqual(job.is_finished, False) self.assertEqual(job.is_finished, False)
self.assertEqual(job.is_failed, True) self.assertEqual(job.is_failed, True)
@ -243,13 +243,13 @@ class TestWorker(RQTestCase):
job = q.enqueue_call(say_hello, depends_on=parent_job) job = q.enqueue_call(say_hello, depends_on=parent_job)
w.work(burst=True) w.work(burst=True)
job = Job.fetch(job.id) job = Job.fetch(job.id)
self.assertEqual(job.status, Status.FINISHED) self.assertEqual(job.get_status(), Status.FINISHED)
parent_job = q.enqueue(div_by_zero) parent_job = q.enqueue(div_by_zero)
job = q.enqueue_call(say_hello, depends_on=parent_job) job = q.enqueue_call(say_hello, depends_on=parent_job)
w.work(burst=True) w.work(burst=True)
job = Job.fetch(job.id) job = Job.fetch(job.id)
self.assertNotEqual(job.status, Status.FINISHED) self.assertNotEqual(job.get_status(), Status.FINISHED)
def test_get_current_job(self): def test_get_current_job(self):
"""Ensure worker.get_current_job() works properly""" """Ensure worker.get_current_job() works properly"""

Write Preview
Loading…
Cancel
Save