Merge branch 'selwin-custom-worker'

rq/job.py

@@ -1,4 +1,3 @@
-import importlib
 import inspect
 import times
 from uuid import uuid4
@@ -9,6 +8,7 @@ except ImportError:  # noqa
 from .local import LocalStack
 from .connections import resolve_connection
 from .exceptions import UnpickleError, NoSuchJobError
+from .utils import import_attribute
 from rq.compat import text_type, decode_redis_hash, as_text
 
 
@@ -151,9 +151,7 @@ class Job(object):
         if self.instance:
             return getattr(self.instance, func_name)
 
-        module_name, func_name = func_name.rsplit('.', 1)
+        return import_attribute(self.func_name)
-        module = importlib.import_module(module_name)
-        return getattr(module, func_name)
 
     @property
     def instance(self):

rq/scripts/rqworker.py

@@ -1,18 +1,16 @@
 #!/usr/bin/env python
-import os
-import sys
 import argparse
 import logging
 import logging.config
+import os
+import sys
 
-from rq import Queue, Worker
+from rq import Queue
 from rq.logutils import setup_loghandlers
 from redis.exceptions import ConnectionError
 from rq.contrib.legacy import cleanup_ghosts
-from rq.scripts import add_standard_arguments
+from rq.scripts import add_standard_arguments, read_config_file, setup_default_arguments, setup_redis
-from rq.scripts import setup_redis
+from rq.utils import import_attribute
-from rq.scripts import read_config_file
-from rq.scripts import setup_default_arguments
 
 logger = logging.getLogger(__name__)
 
@@ -23,6 +21,7 @@ def parse_args():
 
     parser.add_argument('--burst', '-b', action='store_true', default=False, help='Run in burst mode (quit after all work is done)')
     parser.add_argument('--name', '-n', default=None, help='Specify a different name')
+    parser.add_argument('--worker-class', '-w', action='store', default='rq.Worker', help='RQ Worker class to use')
     parser.add_argument('--path', '-P', default='.', help='Specify the import path.')
     parser.add_argument('--verbose', '-v', action='store_true', default=False, help='Show more output')
     parser.add_argument('--quiet', '-q', action='store_true', default=False, help='Show less output')
@@ -75,10 +74,11 @@ def main():
     setup_redis(args)
 
     cleanup_ghosts()
+    worker_class = import_attribute(args.worker_class)
 
     try:
         queues = list(map(Queue, args.queues))
-        w = Worker(queues, name=args.name)
+        w = worker_class(queues, name=args.name)
 
         # Should we configure Sentry?
         if args.sentry_dsn:

rq/utils.py

@@ -5,9 +5,10 @@ Miscellaneous helper functions.
 The formatter for ANSI colored console output is heavily based on Pygments
 terminal colorizing code, originally by Georg Brandl.
 """
+import importlib
+import logging
 import os
 import sys
-import logging
 
 from .compat import is_python_version
 
@@ -160,3 +161,10 @@ class ColorizingStreamHandler(logging.StreamHandler):
             message = '\n'.join(parts)
 
         return message
+
+
+def import_attribute(name):
+    """Return an attribute from a dotted path name (e.g. "path.to.func")."""
+    module_name, attribute = name.rsplit('.', 1)
+    module = importlib.import_module(module_name)
+    return getattr(module, attribute)