mirror of https://github.com/rq/rq.git
commit
e0b26758a2
67
rq/job.py
67
rq/job.py
|
@ -9,11 +9,6 @@ from .connections import resolve_connection
|
|||
from .exceptions import UnpickleError, NoSuchJobError
|
||||
|
||||
|
||||
JOB_ATTRS = set(['origin', '_func_name', 'ended_at', 'description', '_args',
|
||||
'created_at', 'enqueued_at', 'connection', '_result', 'result',
|
||||
'timeout', '_kwargs', 'exc_info', '_id', 'data', '_instance',
|
||||
'result_ttl', '_status', 'status'])
|
||||
|
||||
def enum(name, *sequential, **named):
|
||||
values = dict(zip(sequential, range(len(sequential))), **named)
|
||||
return type(name, (), values)
|
||||
|
@ -167,6 +162,7 @@ class Job(object):
|
|||
self.timeout = None
|
||||
self.result_ttl = None
|
||||
self._status = None
|
||||
self.meta = {}
|
||||
|
||||
|
||||
# Data access
|
||||
|
@ -256,12 +252,8 @@ class Job(object):
|
|||
self.exc_info = obj.get('exc_info')
|
||||
self.timeout = int(obj.get('timeout')) if obj.get('timeout') else None
|
||||
self.result_ttl = int(obj.get('result_ttl')) if obj.get('result_ttl') else None # noqa
|
||||
self._status = obj.get('status') if obj.get('status') else None # noqa
|
||||
|
||||
# Overwrite job's additional attrs (those not in JOB_ATTRS), if any
|
||||
additional_attrs = set(obj.keys()).difference(JOB_ATTRS)
|
||||
for attr in additional_attrs:
|
||||
setattr(self, attr, obj[attr])
|
||||
self._status = obj.get('status') if obj.get('status') else None
|
||||
self.meta = unpickle(obj.get('meta')) if obj.get('meta') else {}
|
||||
|
||||
def save(self):
|
||||
"""Persists the current job instance to its corresponding Redis key."""
|
||||
|
@ -290,18 +282,9 @@ class Job(object):
|
|||
obj['result_ttl'] = self.result_ttl
|
||||
if self._status is not None:
|
||||
obj['status'] = self._status
|
||||
"""
|
||||
Store additional attributes from job instance into Redis. This is done
|
||||
so that third party libraries using RQ can store additional data
|
||||
directly on ``Job`` instances. For example:
|
||||
if self.meta:
|
||||
obj['meta'] = dumps(self.meta)
|
||||
|
||||
job = Job.create(func)
|
||||
job.foo = 'bar'
|
||||
job.save() # Will persist the 'foo' attribute
|
||||
"""
|
||||
additional_attrs = set(self.__dict__.keys()).difference(JOB_ATTRS)
|
||||
for attr in additional_attrs:
|
||||
obj[attr] = getattr(self, attr)
|
||||
self.connection.hmset(key, obj)
|
||||
|
||||
def cancel(self):
|
||||
|
@ -351,3 +334,43 @@ class Job(object):
|
|||
|
||||
def __hash__(self):
|
||||
return hash(self.id)
|
||||
|
||||
|
||||
# Backwards compatibility for custom properties
|
||||
def __getattr__(self, name):
|
||||
import warnings
|
||||
warnings.warn(
|
||||
"Getting custom properties from the job instance directly "
|
||||
"will be unsupported as of RQ 0.4. Please use the meta dict "
|
||||
"to store all custom variables. So instead of this:\n\n"
|
||||
"\tjob.foo\n\n"
|
||||
"Use this:\n\n"
|
||||
"\tjob.meta['foo']\n",
|
||||
SyntaxWarning)
|
||||
try:
|
||||
return self.__dict__['meta'][name] # avoid recursion
|
||||
except KeyError:
|
||||
return getattr(super(Job, self), name)
|
||||
|
||||
def __setattr__(self, name, value):
|
||||
# Ignore the "private" fields
|
||||
private_attrs = set(['origin', '_func_name', 'ended_at',
|
||||
'description', '_args', 'created_at', 'enqueued_at', 'connection',
|
||||
'_result', 'result', 'timeout', '_kwargs', 'exc_info', '_id',
|
||||
'data', '_instance', 'result_ttl', '_status', 'status', 'meta'])
|
||||
|
||||
if name in private_attrs:
|
||||
object.__setattr__(self, name, value)
|
||||
return
|
||||
|
||||
import warnings
|
||||
warnings.warn(
|
||||
"Setting custom properties on the job instance directly will "
|
||||
"be unsupported as of RQ 0.4. Please use the meta dict to "
|
||||
"store all custom variables. So instead of this:\n\n"
|
||||
"\tjob.foo = 'bar'\n\n"
|
||||
"Use this:\n\n"
|
||||
"\tjob.meta['foo'] = 'bar'\n",
|
||||
SyntaxWarning)
|
||||
|
||||
self.__dict__['meta'][name] = value
|
||||
|
|
|
@ -175,20 +175,17 @@ class TestJob(RQTestCase):
|
|||
with self.assertRaises(AttributeError):
|
||||
job.func # accessing the func property should fail
|
||||
|
||||
def test_additional_job_attrs_is_persisted(self):
|
||||
"""
|
||||
Verify that additional attributes stored on jobs are:
|
||||
- Saved in Redis when job.save() is called
|
||||
- Attached back to job instance when job.refresh() is called
|
||||
"""
|
||||
def test_custom_meta_is_persisted(self):
|
||||
"""Additional meta data on jobs are stored persisted correctly."""
|
||||
job = Job.create(func=say_hello, args=('Lionel',))
|
||||
job.foo = 'bar'
|
||||
job.meta['foo'] = 'bar'
|
||||
job.save()
|
||||
self.assertEqual(self.testconn.hget(job.key, 'foo'), 'bar')
|
||||
|
||||
raw_data = self.testconn.hget(job.key, 'meta')
|
||||
self.assertEqual(loads(raw_data)['foo'], 'bar')
|
||||
|
||||
job2 = Job.fetch(job.id)
|
||||
job2.refresh()
|
||||
self.assertEqual(job2.foo, 'bar')
|
||||
self.assertEqual(job2.meta['foo'], 'bar')
|
||||
|
||||
def test_result_ttl_is_persisted(self):
|
||||
"""Ensure that job's result_ttl is set properly"""
|
||||
|
|
Loading…
Reference in New Issue