Merge branch 'job-access-within-job'

Conflicts:
	CHANGES.md
	rq/job.py

CHANGES.md

@@ -1,6 +1,12 @@
 ### 0.3.3
 (not released yet)
 
+- Jobs can now access the current `Job` instance from within.  Relevant
+  documentation [here](http://python-rq.org/docs/jobs/).
+
+- Custom properties can be set by modifying the `job.meta` dict.  Relevant
+  documentation [here](http://python-rq.org/docs/jobs/).
+
 - Custom properties can be set by modifying the `job.meta` dict.  Relevant
   documentation [here](http://python-rq.org/docs/jobs/).
 

rq/__init__.py

@@ -3,6 +3,7 @@ from .connections import use_connection, push_connection, pop_connection
 from .connections import Connection
 from .queue import Queue, get_failed_queue
 from .job import cancel_job, requeue_job
+from .job import get_current_job
 from .worker import Worker
 from .version import VERSION
 
@@ -11,5 +12,5 @@ __all__ = [
     'use_connection', 'get_current_connection',
     'push_connection', 'pop_connection', 'Connection',
     'Queue', 'get_failed_queue', 'Worker',
-    'cancel_job', 'requeue_job']
+    'cancel_job', 'requeue_job', 'get_current_job']
 __version__ = VERSION

rq/job.py

@@ -49,6 +49,16 @@ def requeue_job(job_id, connection=None):
     fq.requeue(job_id)
 
 
+def get_current_job():
+    """Returns the Job instance that is currently being executed.  If this
+    function is invoked from outside a job context, None is returned.
+    """
+    job_id = _job_stack.top
+    if job_id is None:
+        return None
+    return Job.fetch(job_id)
+
+
 class Job(object):
     """A Job is just a convenient datastructure to pass around job (meta) data.
     """
@@ -310,9 +320,12 @@ class Job(object):
 
     # Job execution
     def perform(self):  # noqa
-        """Invokes the job function with the job arguments.
-        """
+        """Invokes the job function with the job arguments."""
+        _job_stack.push(self.id)
+        try:
             self._result = self.func(*self.args, **self.kwargs)
+        finally:
+            assert self.id == _job_stack.pop()
         return self._result
 
 
@@ -342,7 +355,7 @@ class Job(object):
 
 
     # Backwards compatibility for custom properties
-    def __getattr__(self, name):
+    def __getattr__(self, name):  # noqa
         import warnings
         warnings.warn(
                 "Getting custom properties from the job instance directly "
@@ -379,3 +392,6 @@ class Job(object):
                 SyntaxWarning)
 
         self.__dict__['meta'][name] = value
+
+
+_job_stack = LocalStack()

tests/fixtures.py

@@ -5,6 +5,7 @@ fixtures has a slighty different characteristics.
 import time
 from rq import Connection
 from rq.decorators import job
+from rq import get_current_job
 
 
 def say_hello(name=None):
@@ -43,6 +44,11 @@ def create_file_after_timeout(path, timeout):
     create_file(path)
 
 
+def access_self():
+    job = get_current_job()
+    return job.id
+
+
 class Calculator(object):
     """Test instance methods."""
     def __init__(self, denominator):

tests/test_job.py

@@ -1,10 +1,10 @@
 import times
 from datetime import datetime
 from tests import RQTestCase
-from tests.fixtures import Calculator, some_calculation, say_hello
+from tests.fixtures import Calculator, some_calculation, say_hello, access_self
 from tests.helpers import strip_milliseconds
 from cPickle import loads
-from rq.job import Job
+from rq.job import Job, get_current_job
 from rq.exceptions import NoSuchJobError, UnpickleError
 
 
@@ -198,3 +198,16 @@ class TestJob(RQTestCase):
         job.save()
         job_from_queue = Job.fetch(job.id, connection=self.testconn)
         self.assertEqual(job.result_ttl, None)
+
+    def test_job_access_within_job_function(self):
+        """The current job is accessible within the job function."""
+        # Executing the job function from outside of RQ throws an exception
+        self.assertIsNone(get_current_job())
+
+        # Executing the job function from within the job works (and in
+        # this case leads to the job ID being returned)
+        job = Job.create(func=access_self)
+        job.save()
+        id = job.perform()
+        self.assertEqual(job.id, id)
+        self.assertEqual(job.func, access_self)