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

Flake8 style fixes.

Browse Source
main
Vincent Driessen 13 years ago
parent 1a8b80604d
commit 0a0d9d1ceb
6 changed files with 71 additions and 50 deletions
Show Stats Download Patch File Download Diff File

10
rq/dummy.py
Unescape Escape View File

@ -3,12 +3,15 @@ Some dummy tasks that are well-suited for generating load for testing purposes.
""" """
import time import time
def do_nothing(): def do_nothing():
pass pass
def sleep(secs): def sleep(secs):
time.sleep(secs) time.sleep(secs)
def endless_loop(): def endless_loop():
x = 7 x = 7
while True: while True:
@ -18,14 +21,17 @@ def endless_loop():
if x == 0: if x == 0:
x = 82 x = 82
def div_by_zero(): def div_by_zero():
1/0 1 / 0
def fib(n): def fib(n):
if n <= 1: if n <= 1:
return 1 return 1
else: else:
return fib(n-2) + fib(n-1) return fib(n - 2) + fib(n - 1)
def yield_stuff(): def yield_stuff():
yield 7 yield 7

3
rq/exceptions.py
Unescape Escape View File

@ -1,11 +1,12 @@
class NoSuchJobError(Exception): class NoSuchJobError(Exception):
pass pass
class NoQueueError(Exception): class NoQueueError(Exception):
pass pass
class UnpickleError(Exception): class UnpickleError(Exception):
def __init__(self, message, raw_data): def __init__(self, message, raw_data):
super(UnpickleError, self).__init__(message) super(UnpickleError, self).__init__(message)
self.raw_data = raw_data self.raw_data = raw_data

2
rq/proxy.py
Unescape Escape View File

@ -1,5 +1,3 @@
import redis
class NoRedisConnectionException(Exception): class NoRedisConnectionException(Exception):
pass pass

25
rq/queue.py
Unescape Escape View File

@ -69,18 +69,21 @@ class Queue(object):
return conn.llen(self.key) return conn.llen(self.key)
def push_job_id(self, job_id): def push_job_id(self, job_id): # noqa
"""Pushes a job ID on the corresponding Redis queue.""" """Pushes a job ID on the corresponding Redis queue."""
conn.rpush(self.key, job_id) conn.rpush(self.key, job_id)
def enqueue(self, f, *args, **kwargs): def enqueue(self, f, *args, **kwargs):
"""Creates a job to represent the delayed function call and enqueues it. """Creates a job to represent the delayed function call and enqueues
it.
Expects the function to call, along with the arguments and keyword Expects the function to call, along with the arguments and keyword
arguments. arguments.
""" """
if f.__module__ == '__main__': if f.__module__ == '__main__':
raise ValueError('Functions from the __main__ module cannot be processed by workers.') raise ValueError(
'Functions from the __main__ module cannot be processed '
'by workers.')
job = Job.for_call(f, *args, **kwargs) job = Job.for_call(f, *args, **kwargs)
return self.enqueue_job(job) return self.enqueue_job(job)
@ -108,10 +111,10 @@ class Queue(object):
@classmethod @classmethod
def lpop(cls, queue_keys, blocking): def lpop(cls, queue_keys, blocking):
"""Helper method. Intermediate method to abstract away from some Redis """Helper method. Intermediate method to abstract away from some
API details, where LPOP accepts only a single key, whereas BLPOP accepts Redis API details, where LPOP accepts only a single key, whereas BLPOP
multiple. So if we want the non-blocking LPOP, we need to iterate over accepts multiple. So if we want the non-blocking LPOP, we need to
all queues, do individual LPOPs, and return the result. iterate over all queues, do individual LPOPs, and return the result.
Until Redis receives a specific method for this, we'll have to wrap it Until Redis receives a specific method for this, we'll have to wrap it
this way. this way.
@ -149,8 +152,8 @@ class Queue(object):
@classmethod @classmethod
def dequeue_any(cls, queues, blocking): def dequeue_any(cls, queues, blocking):
"""Class method returning the Job instance at the front of the given set """Class method returning the Job instance at the front of the given
of Queues, where the order of the queues is important. set of Queues, where the order of the queues is important.
When all of the Queues are empty, depending on the `blocking` argument, When all of the Queues are empty, depending on the `blocking` argument,
either blocks execution of this function until new messages arrive on either blocks execution of this function until new messages arrive on
@ -179,7 +182,7 @@ class Queue(object):
# Total ordering defition (the rest of the required Python methods are # Total ordering defition (the rest of the required Python methods are
# auto-generated by the @total_ordering decorator) # auto-generated by the @total_ordering decorator)
def __eq__(self, other): def __eq__(self, other): # noqa
if not isinstance(other, Queue): if not isinstance(other, Queue):
raise TypeError('Cannot compare queues to other objects.') raise TypeError('Cannot compare queues to other objects.')
return self.name == other.name return self.name == other.name
@ -193,7 +196,7 @@ class Queue(object):
return hash(self.name) return hash(self.name)
def __repr__(self): def __repr__(self): # noqa
return 'Queue(%r)' % (self.name,) return 'Queue(%r)' % (self.name,)
def __str__(self): def __str__(self):

30
rq/utils.py
Unescape Escape View File

@ -2,15 +2,16 @@
""" """
Miscellaneous helper functions. Miscellaneous helper functions.
The formatter for ANSI colored console output is heavily based on Pygments terminal The formatter for ANSI colored console output is heavily based on Pygments
colorizing code, originally by Georg Brandl. terminal colorizing code, originally by Georg Brandl.
""" """
import os import os
def gettermsize(): def gettermsize():
def ioctl_GWINSZ(fd): def ioctl_GWINSZ(fd):
try: try:
import fcntl, termios, struct import fcntl, termios, struct # noqa
cr = struct.unpack('hh', fcntl.ioctl(fd, termios.TIOCGWINSZ, cr = struct.unpack('hh', fcntl.ioctl(fd, termios.TIOCGWINSZ,
'1234')) '1234'))
except: except:
@ -37,17 +38,17 @@ class _Colorizer(object):
esc = "\x1b[" esc = "\x1b["
self.codes = {} self.codes = {}
self.codes[""] = "" self.codes[""] = ""
self.codes["reset"] = esc + "39;49;00m" self.codes["reset"] = esc + "39;49;00m"
self.codes["bold"] = esc + "01m" self.codes["bold"] = esc + "01m"
self.codes["faint"] = esc + "02m" self.codes["faint"] = esc + "02m"
self.codes["standout"] = esc + "03m" self.codes["standout"] = esc + "03m"
self.codes["underline"] = esc + "04m" self.codes["underline"] = esc + "04m"
self.codes["blink"] = esc + "05m" self.codes["blink"] = esc + "05m"
self.codes["overline"] = esc + "06m" self.codes["overline"] = esc + "06m"
dark_colors = ["black", "darkred", "darkgreen", "brown", "darkblue", dark_colors = ["black", "darkred", "darkgreen", "brown", "darkblue",
"purple", "teal", "lightgray"] "purple", "teal", "lightgray"]
light_colors = ["darkgray", "red", "green", "yellow", "blue", light_colors = ["darkgray", "red", "green", "yellow", "blue",
"fuchsia", "turquoise", "white"] "fuchsia", "turquoise", "white"]
@ -60,10 +61,10 @@ class _Colorizer(object):
del d, l, x del d, l, x
self.codes["darkteal"] = self.codes["turquoise"] self.codes["darkteal"] = self.codes["turquoise"]
self.codes["darkyellow"] = self.codes["brown"] self.codes["darkyellow"] = self.codes["brown"]
self.codes["fuscia"] = self.codes["fuchsia"] self.codes["fuscia"] = self.codes["fuchsia"]
self.codes["white"] = self.codes["bold"] self.codes["white"] = self.codes["bold"]
def reset_color(self): def reset_color(self):
return self.codes["reset"] return self.codes["reset"]
@ -98,6 +99,7 @@ class _Colorizer(object):
colorizer = _Colorizer() colorizer = _Colorizer()
def make_colorizer(color): def make_colorizer(color):
"""Creates a function that colorizes text with the given color. """Creates a function that colorizes text with the given color.

51
rq/worker.py
Unescape Escape View File

@ -11,7 +11,7 @@ import traceback
from cPickle import dumps from cPickle import dumps
try: try:
from logbook import Logger from logbook import Logger
Logger = Logger # Does nothing except it shuts up pyflakes annoying error Logger = Logger # Does nothing except it shuts up pyflakes annoying error
except ImportError: except ImportError:
from logging import Logger from logging import Logger
from .queue import Queue from .queue import Queue
@ -23,9 +23,11 @@ green = make_colorizer('darkgreen')
yellow = make_colorizer('darkyellow') yellow = make_colorizer('darkyellow')
blue = make_colorizer('darkblue') blue = make_colorizer('darkblue')
def iterable(x): def iterable(x):
return hasattr(x, '__iter__') return hasattr(x, '__iter__')
def compact(l): def compact(l):
return [x for x in l if x is not None] return [x for x in l if x is not None]
@ -33,6 +35,7 @@ _signames = dict((getattr(signal, signame), signame) \
for signame in dir(signal) \ for signame in dir(signal) \
if signame.startswith('SIG') and '_' not in signame) if signame.startswith('SIG') and '_' not in signame)
def signal_name(signum): def signal_name(signum):
# Hackety-hack-hack: is there really no better way to reverse lookup the # Hackety-hack-hack: is there really no better way to reverse lookup the
# signal name? If you read this and know a way: please provide a patch :) # signal name? If you read this and know a way: please provide a patch :)
@ -55,9 +58,9 @@ class Worker(object):
@classmethod @classmethod
def find_by_key(cls, worker_key): def find_by_key(cls, worker_key):
"""Returns a Worker instance, based on the naming conventions for naming """Returns a Worker instance, based on the naming conventions for
the internal Redis keys. Can be used to reverse-lookup Workers by their naming the internal Redis keys. Can be used to reverse-lookup Workers
Redis keys. by their Redis keys.
""" """
prefix = cls.redis_worker_namespace_prefix prefix = cls.redis_worker_namespace_prefix
name = worker_key[len(prefix):] name = worker_key[len(prefix):]
@ -76,7 +79,7 @@ class Worker(object):
return worker return worker
def __init__(self, queues, name=None, rv_ttl=500): def __init__(self, queues, name=None, rv_ttl=500): # noqa
if isinstance(queues, Queue): if isinstance(queues, Queue):
queues = [queues] queues = [queues]
self._name = name self._name = name
@ -91,7 +94,7 @@ class Worker(object):
self.failed_queue = Queue('failed') self.failed_queue = Queue('failed')
def validate_queues(self): def validate_queues(self): # noqa
"""Sanity check for the given queues.""" """Sanity check for the given queues."""
if not iterable(self.queues): if not iterable(self.queues):
raise ValueError('Argument queues not iterable.') raise ValueError('Argument queues not iterable.')
@ -108,7 +111,7 @@ class Worker(object):
return map(lambda q: q.key, self.queues) return map(lambda q: q.key, self.queues)
@property @property # noqa
def name(self): def name(self):
"""Returns the name of the worker, under which it is registered to the """Returns the name of the worker, under which it is registered to the
monitoring system. monitoring system.
@ -152,11 +155,13 @@ class Worker(object):
procname.setprocname('rq: %s' % (message,)) procname.setprocname('rq: %s' % (message,))
def register_birth(self): def register_birth(self): # noqa
"""Registers its own birth.""" """Registers its own birth."""
self.log.debug('Registering birth of worker %s' % (self.name,)) self.log.debug('Registering birth of worker %s' % (self.name,))
if conn.exists(self.key) and not conn.hexists(self.key, 'death'): if conn.exists(self.key) and not conn.hexists(self.key, 'death'):
raise ValueError('There exists an active worker named \'%s\' alread.' % (self.name,)) raise ValueError(
'There exists an active worker named \'%s\' '
'already.' % (self.name,))
key = self.key key = self.key
now = time.time() now = time.time()
queues = ','.join(self.queue_names()) queues = ','.join(self.queue_names())
@ -203,7 +208,8 @@ class Worker(object):
# Take down the horse with the worker # Take down the horse with the worker
if self.horse_pid: if self.horse_pid:
self.log.debug('Taking down horse %d with me.' % self.horse_pid) msg = 'Taking down horse %d with me.' % self.horse_pid
self.log.debug(msg)
try: try:
os.kill(self.horse_pid, signal.SIGKILL) os.kill(self.horse_pid, signal.SIGKILL)
except OSError as e: except OSError as e:
@ -226,7 +232,8 @@ class Worker(object):
self.log.debug('Ignoring signal %s.' % signal_name(signum)) self.log.debug('Ignoring signal %s.' % signal_name(signum))
return return
self.log.warning('Warm shut down. Press Ctrl+C again for a cold shutdown.') msg = 'Warm shut down. Press Ctrl+C again for a cold shutdown.'
self.log.warning(msg)
self._stopped = True self._stopped = True
self.log.debug('Stopping after current horse is finished.') self.log.debug('Stopping after current horse is finished.')
@ -234,7 +241,7 @@ class Worker(object):
signal.signal(signal.SIGTERM, request_stop) signal.signal(signal.SIGTERM, request_stop)
def work(self, burst=False): def work(self, burst=False): # noqa
"""Starts the work loop. """Starts the work loop.
Pops and performs all jobs on the current list of queues. When all Pops and performs all jobs on the current list of queues. When all
@ -257,14 +264,17 @@ class Worker(object):
qnames = self.queue_names() qnames = self.queue_names()
self.procline('Listening on %s' % ','.join(qnames)) self.procline('Listening on %s' % ','.join(qnames))
self.log.info('') self.log.info('')
self.log.info('*** Listening on %s...' % (green(', '.join(qnames)))) self.log.info('*** Listening on %s...' % \
green(', '.join(qnames)))
wait_for_job = not burst wait_for_job = not burst
try: try:
result = Queue.dequeue_any(self.queues, wait_for_job) result = Queue.dequeue_any(self.queues, wait_for_job)
if result is None: if result is None:
break break
except UnpickleError as e: except UnpickleError as e:
self.log.warning('*** Ignoring unpickleable data on %s.' % (e.queue.name,)) msg = '*** Ignoring unpickleable data on %s.' % \
green(e.queue.name)
self.log.warning(msg)
self.log.debug('Data follows:') self.log.debug('Data follows:')
self.log.debug(e.raw_data) self.log.debug(e.raw_data)
self.log.debug('End of unreadable data.') self.log.debug('End of unreadable data.')
@ -272,7 +282,8 @@ class Worker(object):
continue continue
job, queue = result job, queue = result
self.log.info('%s: %s (%s)' % (green(queue.name), blue(job.description), job.id)) self.log.info('%s: %s (%s)' % (green(queue.name),
blue(job.description), job.id))
self.state = 'busy' self.state = 'busy'
self.fork_and_perform_job(job) self.fork_and_perform_job(job)
@ -301,11 +312,11 @@ class Worker(object):
break break
except OSError as e: except OSError as e:
# In case we encountered an OSError due to EINTR (which is # In case we encountered an OSError due to EINTR (which is
# caused by a SIGINT or SIGTERM signal during os.waitpid()), # caused by a SIGINT or SIGTERM signal during
# we simply ignore it and enter the next iteration of the # os.waitpid()), we simply ignore it and enter the next
# loop, waiting for the child to end. In any other case, # iteration of the loop, waiting for the child to end. In
# this is some other unexpected OS error, which we don't # any other case, this is some other unexpected OS error,
# want to catch, so we re-raise those ones. # which we don't want to catch, so we re-raise those ones.
if e.errno != errno.EINTR: if e.errno != errno.EINTR:
raise raise

Write Preview
Loading…
Cancel
Save