New connection management.

Connections can now be set explicitly on Queues, Workers, and Jobs.
Jobs that are implicitly created by Queue or Worker API calls now
inherit the connection of their creator's.

For all RQ object instances that are created now holds that the
"current" connection is used if none is passed in explicitly.  The
"current" connection is thus hold on to at creation time and won't be
changed for the lifetime of the object.

Effectively, this means that, given a default Redis connection, say you
create a queue Q1, then push another Redis connection onto the
connection stack, then create Q2. In that case, Q1 means a queue on the
first connection and Q2 on the second connection.

This is way more clear than it used to be.

Also, I've removed the `use_redis()` call, which was named ugly.
Instead, some new alternatives for connection management now exist.

You can push/pop connections now:

    >>> my_conn = Redis()
    >>> push_connection(my_conn)
    >>> q = Queue()
    >>> q.connection == my_conn
    True
    >>> pop_connection() == my_conn

Also, you can stack them syntactically:

    >>> conn1 = Redis()
    >>> conn2 = Redis('example.org', 1234)
    >>> with Connection(conn1):
    ...     q = Queue()
    ...     with Connection(conn2):
    ...         q2 = Queue()
    ...     q3 = Queue()
    >>> q.connection == conn1
    True
    >>> q2.connection == conn2
    True
    >>> q3.connection == conn1
    True

Or, if you only require a single connection to Redis (for most uses):

    >>> use_connection(Redis())

rq/__init__.py

@@ -1,27 +1,19 @@
-from redis import Redis
-from .proxy import conn
+from .connections import get_current_connection
+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 .worker import Worker
 from .version import VERSION
 
 
 def use_redis(redis=None):
-    """Pushes the given Redis connection (a redis.Redis instance) onto the
-    connection stack.  This is merely a helper function to easily start
-    using RQ without having to know or understand the RQ connection stack.
+    use_connection(redis)
 
-    When given None as an argument, a (default) Redis connection to
-    redis://localhost:6379 is set up.
-    """
-    if redis is None:
-        redis = Redis()
-    elif not isinstance(redis, Redis):
-        raise TypeError('Argument redis should be a Redis instance.')
-    conn.push(redis)
 
 __all__ = [
-    'conn', 'use_redis',
+    'use_connection', 'get_current_connection',
+    'push_connection', 'pop_connection', 'Connection',
     'Queue', 'get_failed_queue', 'Worker',
     'cancel_job', 'requeue_job']
 __version__ = VERSION

rq/connections.py

@@ -0,0 +1,82 @@
+from contextlib import contextmanager
+from redis import Redis
+
+
+class NoRedisConnectionException(Exception):
+    pass
+
+
+class _RedisConnectionStack(object):
+    def __init__(self):
+        self.stack = []
+
+    def _get_current_object(self):
+        try:
+            return self.stack[-1]
+        except IndexError:
+            msg = 'No Redis connection configured.'
+            raise NoRedisConnectionException(msg)
+
+    def pop(self):
+        return self.stack.pop()
+
+    def push(self, connection):
+        self.stack.append(connection)
+
+    def empty(self):
+        del self.stack[:]
+
+    def depth(self):
+        return len(self.stack)
+
+    def __getattr__(self, name):
+        return getattr(self._get_current_object(), name)
+
+
+_connection_stack = _RedisConnectionStack()
+
+
+@contextmanager
+def Connection(connection=None):
+    if connection is None:
+        connection = Redis()
+    _connection_stack.push(connection)
+    try:
+        yield
+    finally:
+        popped = _connection_stack.pop()
+        assert popped == connection, \
+                'Unexpected Redis connection was popped off the stack. ' \
+                'Check your Redis connection setup.'
+
+
+def push_connection(redis):
+    """Pushes the given connection on the stack."""
+    _connection_stack.push(redis)
+
+
+def pop_connection():
+    """Pops the topmost connection from the stack."""
+    return _connection_stack.pop()
+
+
+def use_connection(redis):
+    """Clears the stack and uses the given connection.  Protects against mixed
+    use of use_connection() and stacked connection contexts.
+    """
+    assert _connection_stack.depth() <= 1, \
+            'You should not mix Connection contexts with use_connection().'
+    _connection_stack.empty()
+    push_connection(redis)
+
+
+def get_current_connection():
+    """Returns the current Redis connection (i.e. the topmost on the
+    connection stack).
+    """
+    return _connection_stack._get_current_object()
+
+
+__all__ = ['Connection',
+        'get_current_connection', 'push_connection', 'pop_connection',
+        'use_connection']

rq/job.py

@@ -2,7 +2,7 @@
 import times
 from uuid import uuid4
 from cPickle import loads, dumps, UnpicklingError
-from .proxy import conn
+from .connections import get_current_connection
 from .exceptions import UnpickleError, NoSuchJobError
 
 
@@ -21,20 +21,20 @@ def unpickle(pickled_string):
     return obj
 
 
-def cancel_job(job_id):
+def cancel_job(job_id, connection=None):
     """Cancels the job with the given job ID, preventing execution.  Discards
     any job info (i.e. it can't be requeued later).
     """
-    Job(job_id).cancel()
+    Job(job_id, connection=connection).cancel()
 
 
-def requeue_job(job_id):
+def requeue_job(job_id, connection=None):
     """Requeues the job with the given job ID.  The job ID should refer to
     a failed job (i.e. it should be on the failed queue).  If no such (failed)
     job exists, a NoSuchJobError is raised.
     """
     from .queue import get_failed_queue
-    fq = get_failed_queue()
+    fq = get_failed_queue(connection=connection)
     fq.requeue(job_id)
 
 
@@ -48,7 +48,8 @@ def create(cls, func, *args, **kwargs):
         """Creates a new Job instance for the given function, arguments, and
         keyword arguments.
         """
-        job = Job()
+        connection = kwargs.pop('connection', None)
+        job = Job(connection=connection)
         job._func_name = '%s.%s' % (func.__module__, func.__name__)
         job._args = args
         job._kwargs = kwargs
@@ -80,18 +81,22 @@ def kwargs(self):
     @classmethod
     def exists(cls, job_id):
         """Returns whether a job hash exists for the given job ID."""
+        conn = get_current_connection()
         return conn.exists(cls.key_for(job_id))
 
     @classmethod
-    def fetch(cls, id):
+    def fetch(cls, id, connection=None):
         """Fetches a persisted job from its corresponding Redis key and
         instantiates it.
         """
-        job = Job(id)
+        job = Job(id, connection=connection)
         job.refresh()
         return job
 
-    def __init__(self, id=None):
+    def __init__(self, id=None, connection=None):
+        if connection is None:
+            connection = get_current_connection()
+        self.connection = connection
         self._id = id
         self.created_at = times.now()
         self._func_name = None
@@ -156,7 +161,7 @@ def return_value(self):
         seconds by default).
         """
         if self._result is None:
-            rv = conn.hget(self.key, 'result')
+            rv = self.connection.hget(self.key, 'result')
             if rv is not None:
                 # cache the result
                 self._result = loads(rv)
@@ -175,7 +180,7 @@ def refresh(self):  # noqa
                 'enqueued_at', 'ended_at', 'result', 'exc_info', 'timeout']
         data, created_at, origin, description, \
                 enqueued_at, ended_at, result, \
-                exc_info, timeout = conn.hmget(key, properties)
+                exc_info, timeout = self.connection.hmget(key, properties)
         if data is None:
             raise NoSuchJobError('No such job: %s' % (key,))
 
@@ -222,7 +227,7 @@ def save(self):
         if self.timeout is not None:
             obj['timeout'] = self.timeout
 
-        conn.hmset(key, obj)
+        self.connection.hmset(key, obj)
 
     def cancel(self):
         """Cancels the given job, which will prevent the job from ever being
@@ -237,7 +242,7 @@ def cancel(self):
 
     def delete(self):
         """Deletes the job hash from Redis."""
-        conn.delete(self.key)
+        self.connection.delete(self.key)
 
 
     # Job execution