diff --git a/django_q/cluster.py b/django_q/cluster.py
index bece2df..c6bffae 100644
--- a/django_q/cluster.py
+++ b/django_q/cluster.py
@@ -4,27 +4,30 @@ from __future__ import division
from __future__ import print_function
from __future__ import unicode_literals
-from time import sleep
-
-# external
-import arrow
import ast
+
# Standard
import importlib
import signal
import socket
import traceback
import uuid
+from multiprocessing import Event, Process, Value, current_process
+from time import sleep
+
+# external
+import arrow
+
# Django
from django import db
from django.utils import timezone
from django.utils.translation import gettext_lazy as _
-from multiprocessing import Event, Process, Value, current_process
# Local
import django_q.tasks
from django_q.brokers import get_broker
from django_q.conf import Conf, logger, psutil, get_ppid, error_reporter
+from django_q.humanhash import humanize
from django_q.models import Task, Success, Schedule
from django_q.queues import Queue
from django_q.signals import pre_execute
@@ -49,10 +52,18 @@ class Cluster(object):
# Start Sentinel
self.stop_event = Event()
self.start_event = Event()
- self.sentinel = Process(target=Sentinel,
- args=(self.stop_event, self.start_event, self.cluster_id, self.broker, self.timeout))
+ self.sentinel = Process(
+ target=Sentinel,
+ args=(
+ self.stop_event,
+ self.start_event,
+ self.cluster_id,
+ self.broker,
+ self.timeout,
+ ),
+ )
self.sentinel.start()
- logger.info(_('Q Cluster-{} starting.').format(self.cluster_id))
+ logger.info(_(f"Q Cluster {self.name} starting."))
while not self.start_event.is_set():
sleep(0.1)
return self.pid
@@ -60,17 +71,20 @@ class Cluster(object):
def stop(self):
if not self.sentinel.is_alive():
return False
- logger.info(_('Q Cluster-{} stopping.').format(self.cluster_id))
+ logger.info(_(f"Q Cluster {self.name} stopping."))
self.stop_event.set()
self.sentinel.join()
- logger.info(_('Q Cluster-{} has stopped.').format(self.cluster_id))
+ logger.info(_(f"Q Cluster {self.name} has stopped."))
self.start_event = None
self.stop_event = None
return True
def sig_handler(self, signum, frame):
- logger.debug(_('{} got signal {}').format(current_process().name,
- Conf.SIGNAL_NAMES.get(signum, 'UNKNOWN')))
+ logger.debug(
+ _(
+ f'{current_process().name} got signal {Conf.SIGNAL_NAMES.get(signum, "UNKNOWN")}'
+ )
+ )
self.stop()
@property
@@ -79,6 +93,10 @@ class Cluster(object):
return Stat.get(pid=self.pid, cluster_id=self.cluster_id)
return Status(pid=self.pid, cluster_id=self.cluster_id)
+ @property
+ def name(self):
+ return humanize(self.cluster_id.hex)
+
@property
def is_starting(self):
return self.stop_event and self.start_event and not self.start_event.is_set()
@@ -89,7 +107,12 @@ class Cluster(object):
@property
def is_stopping(self):
- return self.stop_event and self.start_event and self.start_event.is_set() and self.stop_event.is_set()
+ return (
+ self.stop_event
+ and self.start_event
+ and self.start_event.is_set()
+ and self.stop_event.is_set()
+ )
@property
def has_stopped(self):
@@ -97,7 +120,15 @@ class Cluster(object):
class Sentinel(object):
- def __init__(self, stop_event, start_event, cluster_id, broker=None, timeout=Conf.TIMEOUT, start=True):
+ def __init__(
+ self,
+ stop_event,
+ start_event,
+ cluster_id,
+ broker=None,
+ timeout=Conf.TIMEOUT,
+ start=True,
+ ):
# Make sure we catch signals for the pool
signal.signal(signal.SIGINT, signal.SIG_IGN)
signal.signal(signal.SIGTERM, signal.SIG_DFL)
@@ -113,7 +144,9 @@ class Sentinel(object):
self.pool_size = Conf.WORKERS
self.pool = []
self.timeout = timeout
- self.task_queue = Queue(maxsize=Conf.QUEUE_LIMIT) if Conf.QUEUE_LIMIT else Queue()
+ self.task_queue = (
+ Queue(maxsize=Conf.QUEUE_LIMIT) if Conf.QUEUE_LIMIT else Queue()
+ )
self.result_queue = Queue()
self.event_out = Event()
self.monitor = None
@@ -155,7 +188,9 @@ class Sentinel(object):
return self.spawn_process(pusher, self.task_queue, self.event_out, self.broker)
def spawn_worker(self):
- self.spawn_process(worker, self.task_queue, self.result_queue, Value('f', -1), self.timeout)
+ self.spawn_process(
+ worker, self.task_queue, self.result_queue, Value("f", -1), self.timeout
+ )
def spawn_monitor(self):
return self.spawn_process(monitor, self.result_queue, self.broker)
@@ -168,21 +203,21 @@ class Sentinel(object):
close_old_django_connections()
if process == self.monitor:
self.monitor = self.spawn_monitor()
- logger.error(_("reincarnated monitor {} after sudden death").format(process.name))
+ logger.error(_(f"reincarnated monitor {process.name} after sudden death"))
elif process == self.pusher:
self.pusher = self.spawn_pusher()
- logger.error(_("reincarnated pusher {} after sudden death").format(process.name))
+ logger.error(_(f"reincarnated pusher {process.name} after sudden death"))
else:
self.pool.remove(process)
self.spawn_worker()
if process.timer.value == 0:
# only need to terminate on timeout, otherwise we risk destabilizing the queues
process.terminate()
- logger.warn(_("reincarnated worker {} after timeout").format(process.name))
+ logger.warn(_(f"reincarnated worker {process.name} after timeout"))
elif int(process.timer.value) == -2:
- logger.info(_("recycled worker {}").format(process.name))
+ logger.info(_(f"recycled worker {process.name}"))
else:
- logger.error(_("reincarnated worker {} after death").format(process.name))
+ logger.error(_(f"reincarnated worker {process.name} after death"))
self.reincarnations += 1
@@ -201,10 +236,14 @@ class Sentinel(object):
set_cpu_affinity(Conf.CPU_AFFINITY, [w.pid for w in self.pool])
def guard(self):
- logger.info(_('{} guarding cluster at {}').format(current_process().name, self.pid))
+ logger.info(
+ _(
+ f"{current_process().name} guarding cluster {humanize(self.cluster_id.hex)}"
+ )
+ )
self.start_event.set()
Stat(self).save()
- logger.info(_('Q Cluster-{} running.').format(self.parent_pid))
+ logger.info(_(f"Q Cluster {humanize(self.cluster_id.hex)} running."))
counter = 0
cycle = Conf.GUARD_CYCLE # guard loop sleep in seconds
# Guard loop. Runs at least once
@@ -238,7 +277,7 @@ class Sentinel(object):
def stop(self):
Stat(self).save()
name = current_process().name
- logger.info(_('{} stopping cluster processes').format(name))
+ logger.info(_(f"{name} stopping cluster processes"))
# Stopping pusher
self.event_out.set()
# Wait for it to stop
@@ -247,7 +286,7 @@ class Sentinel(object):
Stat(self).save()
# Put poison pills in the queue
for __ in range(len(self.pool)):
- self.task_queue.put('STOP')
+ self.task_queue.put("STOP")
self.task_queue.close()
# wait for the task queue to empty
self.task_queue.join_thread()
@@ -259,11 +298,11 @@ class Sentinel(object):
sleep(0.1)
Stat(self).save()
# Finally stop the monitor
- self.result_queue.put('STOP')
+ self.result_queue.put("STOP")
self.result_queue.close()
# Wait for the result queue to empty
self.result_queue.join_thread()
- logger.info(_('{} waiting for the monitor.').format(name))
+ logger.info(_(f"{name} waiting for the monitor."))
# Wait for everything to close or time out
count = 0
if not self.timeout:
@@ -284,7 +323,7 @@ def pusher(task_queue, event, broker=None):
"""
if not broker:
broker = get_broker()
- logger.info(_('{} pushing tasks at {}').format(current_process().name, current_process().pid))
+ logger.info(_(f"{current_process().name} pushing tasks at {current_process().pid}"))
while True:
try:
task_set = broker.dequeue()
@@ -303,12 +342,12 @@ def pusher(task_queue, event, broker=None):
logger.error(e, traceback.format_exc())
broker.fail(ack_id)
continue
- task['ack_id'] = ack_id
+ task["ack_id"] = ack_id
task_queue.put(task)
- logger.debug(_('queueing from {}').format(broker.list_key))
+ logger.debug(_(f"queueing from {broker.list_key}"))
if event.is_set():
break
- logger.info(_("{} stopped pushing tasks").format(current_process().name))
+ logger.info(_(f"{current_process().name} stopped pushing tasks"))
def monitor(result_queue, broker=None):
@@ -319,25 +358,25 @@ def monitor(result_queue, broker=None):
if not broker:
broker = get_broker()
name = current_process().name
- logger.info(_("{} monitoring at {}").format(name, current_process().pid))
- for task in iter(result_queue.get, 'STOP'):
+ logger.info(_(f"{name} monitoring at {current_process().pid}"))
+ for task in iter(result_queue.get, "STOP"):
# save the result
- if task.get('cached', False):
+ if task.get("cached", False):
save_cached(task, broker)
else:
save_task(task, broker)
# acknowledge result
- ack_id = task.pop('ack_id', False)
- if ack_id and (task['success'] or task.get('ack_failure', False)):
+ ack_id = task.pop("ack_id", False)
+ if ack_id and (task["success"] or task.get("ack_failure", False)):
broker.acknowledge(ack_id)
# log the result
- if task['success']:
+ if task["success"]:
# log success
- logger.info(_("Processed [{}]").format(task['name']))
+ logger.info(_(f"Processed [{task['name']}]"))
else:
# log failure
- logger.error(_("Failed [{}] - {}").format(task['name'], task['result']))
- logger.info(_("{} stopped monitoring results").format(name))
+ logger.error(_(f"Failed [{task['name']}] - {task['result']}"))
+ logger.info(_(f"{name} stopped monitoring results"))
def worker(task_queue, result_queue, timer, timeout=Conf.TIMEOUT):
@@ -348,22 +387,22 @@ def worker(task_queue, result_queue, timer, timeout=Conf.TIMEOUT):
:type timer: multiprocessing.Value
"""
name = current_process().name
- logger.info(_('{} ready for work at {}').format(name, current_process().pid))
+ logger.info(_(f"{name} ready for work at {current_process().pid}"))
task_count = 0
if timeout is None:
timeout = -1
# Start reading the task queue
- for task in iter(task_queue.get, 'STOP'):
+ for task in iter(task_queue.get, "STOP"):
result = None
timer.value = -1 # Idle
task_count += 1
# Get the function from the task
- logger.info(_('{} processing [{}]').format(name, task['name']))
- f = task['func']
+ logger.info(_(f'{name} processing [{task["name"]}]'))
+ f = task["func"]
# if it's not an instance try to get it from the string
- if not callable(task['func']):
+ if not callable(task["func"]):
try:
- module, func = f.rsplit('.', 1)
+ module, func = f.rsplit(".", 1)
m = importlib.import_module(module)
f = getattr(m, func)
except (ValueError, ImportError, AttributeError) as e:
@@ -373,30 +412,30 @@ def worker(task_queue, result_queue, timer, timeout=Conf.TIMEOUT):
# We're still going
if not result:
close_old_django_connections()
- timer_value = task.pop('timeout', timeout)
+ timer_value = task.pop("timeout", timeout)
# signal execution
pre_execute.send(sender="django_q", func=f, task=task)
# execute the payload
timer.value = timer_value # Busy
try:
- res = f(*task['args'], **task['kwargs'])
+ res = f(*task["args"], **task["kwargs"])
result = (res, True)
except Exception as e:
- result = ('{} : {}'.format(e, traceback.format_exc()), False)
+ result = (f"{e} : {traceback.format_exc()}", False)
if error_reporter:
error_reporter.report()
with timer.get_lock():
# Process result
- task['result'] = result[0]
- task['success'] = result[1]
- task['stopped'] = timezone.now()
+ task["result"] = result[0]
+ task["success"] = result[1]
+ task["stopped"] = timezone.now()
result_queue.put(task)
timer.value = -1 # Idle
# Recycle
if task_count == Conf.RECYCLE:
timer.value = -2 # Recycled
break
- logger.info(_('{} stopped doing work').format(name))
+ logger.info(_(f"{name} stopped doing work"))
def save_task(task, broker):
@@ -404,67 +443,77 @@ def save_task(task, broker):
Saves the task package to Django or the cache
"""
# SAVE LIMIT < 0 : Don't save success
- if not task.get('save', Conf.SAVE_LIMIT >= 0) and task['success']:
+ if not task.get("save", Conf.SAVE_LIMIT >= 0) and task["success"]:
return
# enqueues next in a chain
- if task.get('chain', None):
- django_q.tasks.async_chain(task['chain'], group=task['group'], cached=task['cached'], sync=task['sync'], broker=broker)
+ if task.get("chain", None):
+ django_q.tasks.async_chain(
+ task["chain"],
+ group=task["group"],
+ cached=task["cached"],
+ sync=task["sync"],
+ broker=broker,
+ )
# SAVE LIMIT > 0: Prune database, SAVE_LIMIT 0: No pruning
close_old_django_connections()
try:
- if task['success'] and 0 < Conf.SAVE_LIMIT <= Success.objects.count():
+ if task["success"] and 0 < Conf.SAVE_LIMIT <= Success.objects.count():
Success.objects.last().delete()
# check if this task has previous results
- if Task.objects.filter(id=task['id'], name=task['name']).exists():
- existing_task = Task.objects.get(id=task['id'], name=task['name'])
+ if Task.objects.filter(id=task["id"], name=task["name"]).exists():
+ existing_task = Task.objects.get(id=task["id"], name=task["name"])
# only update the result if it hasn't succeeded yet
if not existing_task.success:
- existing_task.stopped = task['stopped']
- existing_task.result = task['result']
- existing_task.success = task['success']
+ existing_task.stopped = task["stopped"]
+ existing_task.result = task["result"]
+ existing_task.success = task["success"]
existing_task.save()
else:
- Task.objects.create(id=task['id'],
- name=task['name'],
- func=task['func'],
- hook=task.get('hook'),
- args=task['args'],
- kwargs=task['kwargs'],
- started=task['started'],
- stopped=task['stopped'],
- result=task['result'],
- group=task.get('group'),
- success=task['success']
- )
+ Task.objects.create(
+ id=task["id"],
+ name=task["name"],
+ func=task["func"],
+ hook=task.get("hook"),
+ args=task["args"],
+ kwargs=task["kwargs"],
+ started=task["started"],
+ stopped=task["stopped"],
+ result=task["result"],
+ group=task.get("group"),
+ success=task["success"],
+ )
except Exception as e:
logger.error(e)
def save_cached(task, broker):
- task_key = '{}:{}'.format(broker.list_key, task['id'])
- timeout = task['cached']
+ task_key = f'{broker.list_key}:{task["id"]}'
+ timeout = task["cached"]
if timeout is True:
timeout = None
try:
- group = task.get('group', None)
- iter_count = task.get('iter_count', 0)
+ group = task.get("group", None)
+ iter_count = task.get("iter_count", 0)
# if it's a group append to the group list
if group:
- group_key = '{}:{}:keys'.format(broker.list_key, group)
+ group_key = f"{broker.list_key}:{group}:keys"
group_list = broker.cache.get(group_key) or []
# if it's an iter group, check if we are ready
if iter_count and len(group_list) == iter_count - 1:
- group_args = '{}:{}:args'.format(broker.list_key, group)
+ group_args = f"{broker.list_key}:{group}:args"
# collate the results into a Task result
- results = [SignedPackage.loads(broker.cache.get(k))['result'] for k in group_list]
- results.append(task['result'])
- task['result'] = results
- task['id'] = group
- task['args'] = SignedPackage.loads(broker.cache.get(group_args))
- task.pop('iter_count', None)
- task.pop('group', None)
- if task.get('iter_cached', None):
- task['cached'] = task.pop('iter_cached', None)
+ results = [
+ SignedPackage.loads(broker.cache.get(k))["result"]
+ for k in group_list
+ ]
+ results.append(task["result"])
+ task["result"] = results
+ task["id"] = group
+ task["args"] = SignedPackage.loads(broker.cache.get(group_args))
+ task.pop("iter_count", None)
+ task.pop("group", None)
+ if task.get("iter_cached", None):
+ task["cached"] = task.pop("iter_cached", None)
save_cached(task, broker=broker)
else:
save_task(task, broker)
@@ -475,12 +524,16 @@ def save_cached(task, broker):
group_list.append(task_key)
broker.cache.set(group_key, group_list, timeout)
# async_task next in a chain
- if task.get('chain', None):
- django_q.tasks.async_chain(task['chain'], group=group, cached=task['cached'], sync=task['sync'], broker=broker)
+ if task.get("chain", None):
+ django_q.tasks.async_chain(
+ task["chain"],
+ group=group,
+ cached=task["cached"],
+ sync=task["sync"],
+ broker=broker,
+ )
# save the task
- broker.cache.set(task_key,
- SignedPackage.dumps(task),
- timeout)
+ broker.cache.set(task_key, SignedPackage.dumps(task), timeout)
except Exception as e:
logger.error(e)
@@ -494,14 +547,18 @@ def scheduler(broker=None):
close_old_django_connections()
try:
with db.transaction.atomic():
- for s in Schedule.objects.select_for_update().exclude(repeats=0).filter(next_run__lt=timezone.now()):
+ for s in (
+ Schedule.objects.select_for_update()
+ .exclude(repeats=0)
+ .filter(next_run__lt=timezone.now())
+ ):
args = ()
kwargs = {}
# get args, kwargs and hook
if s.kwargs:
try:
# eval should be safe here because dict()
- kwargs = eval('dict({})'.format(s.kwargs))
+ kwargs = eval(f"dict({s.kwargs})")
except SyntaxError:
kwargs = {}
if s.args:
@@ -509,9 +566,9 @@ def scheduler(broker=None):
# single value won't eval to tuple, so:
if type(args) != tuple:
args = (args,)
- q_options = kwargs.get('q_options', {})
+ q_options = kwargs.get("q_options", {})
if s.hook:
- q_options['hook'] = s.hook
+ q_options["hook"] = s.hook
# set up the next run time
if not s.schedule_type == s.ONCE:
next_run = arrow.get(s.next_run)
@@ -535,18 +592,23 @@ def scheduler(broker=None):
s.next_run = next_run.datetime
s.repeats += -1
# send it to the cluster
- q_options['broker'] = broker
- q_options['group'] = q_options.get('group', s.name or s.id)
- kwargs['q_options'] = q_options
+ q_options["broker"] = broker
+ q_options["group"] = q_options.get("group", s.name or s.id)
+ kwargs["q_options"] = q_options
s.task = django_q.tasks.async_task(s.func, *args, **kwargs)
# log it
if not s.task:
logger.error(
- _('{} failed to create a task from schedule [{}]').format(current_process().name,
- s.name or s.id))
+ _(
+ f"{current_process().name} failed to create a task from schedule [{s.name or s.id}]"
+ )
+ )
else:
logger.info(
- _('{} created a task from schedule [{}]').format(current_process().name, s.name or s.id))
+ _(
+ f"{current_process().name} created a task from schedule [{s.name or s.id}]"
+ )
+ )
# default behavior is to delete a ONCE schedule
if s.schedule_type == s.ONCE:
if s.repeats < 0:
@@ -561,14 +623,15 @@ def scheduler(broker=None):
def close_old_django_connections():
- '''
+ """
Close django connections unless running with sync=True.
- '''
+ """
if Conf.SYNC:
logger.warning(
- 'Preserving django database connections because sync=True. Beware '
- 'that tasks are now injected in the calling context/transactions '
- 'which may result in unexpected bahaviour.')
+ "Preserving django database connections because sync=True. Beware "
+ "that tasks are now injected in the calling context/transactions "
+ "which may result in unexpected bahaviour."
+ )
else:
db.close_old_connections()
@@ -583,11 +646,13 @@ def set_cpu_affinity(n, process_ids, actual=not Conf.TESTING):
"""
# check if we have the psutil module
if not psutil:
- logger.warning('Skipping cpu affinity because psutil was not found.')
+ logger.warning("Skipping cpu affinity because psutil was not found.")
return
# check if the platform supports cpu_affinity
- if actual and not hasattr(psutil.Process(process_ids[0]), 'cpu_affinity'):
- logger.warning('Faking cpu affinity because it is not supported on this platform')
+ if actual and not hasattr(psutil.Process(process_ids[0]), "cpu_affinity"):
+ logger.warning(
+ "Faking cpu affinity because it is not supported on this platform"
+ )
actual = False
# get the available processors
cpu_list = list(range(psutil.cpu_count()))
@@ -607,4 +672,4 @@ def set_cpu_affinity(n, process_ids, actual=not Conf.TESTING):
p = psutil.Process(pid)
if actual:
p.cpu_affinity(affinity)
- logger.info(_('{} will use cpu {}').format(pid, affinity))
+ logger.info(_(f"{pid} will use cpu {affinity}"))
diff --git a/django_q/conf.py b/django_q/conf.py
index 1c11160..baa4671 100644
--- a/django_q/conf.py
+++ b/django_q/conf.py
@@ -1,15 +1,16 @@
import logging
-from copy import deepcopy
-from signal import signal
-from multiprocessing import cpu_count
-
-# django
-from django.utils.translation import gettext_lazy as _
-from django.conf import settings
# external
import os
+from copy import deepcopy
+from multiprocessing import cpu_count
+from signal import signal
+
import pkg_resources
+from django.conf import settings
+
+# django
+from django.utils.translation import gettext_lazy as _
# local
from django_q.queues import Queue
@@ -25,64 +26,65 @@ class Conf(object):
"""
Configuration class
"""
+
try:
conf = settings.Q_CLUSTER
except AttributeError:
conf = {}
# Redis server configuration . Follows standard redis keywords
- REDIS = conf.get('redis', {})
+ REDIS = conf.get("redis", {})
# Support for Django-Redis connections
- DJANGO_REDIS = conf.get('django_redis', None)
+ DJANGO_REDIS = conf.get("django_redis", None)
# Disque broker
- DISQUE_NODES = conf.get('disque_nodes', None)
+ DISQUE_NODES = conf.get("disque_nodes", None)
# Optional Authentication
- DISQUE_AUTH = conf.get('disque_auth', None)
+ DISQUE_AUTH = conf.get("disque_auth", None)
# Optional Fast acknowledge
- DISQUE_FASTACK = conf.get('disque_fastack', False)
+ DISQUE_FASTACK = conf.get("disque_fastack", False)
# IronMQ broker
- IRON_MQ = conf.get('iron_mq', None)
+ IRON_MQ = conf.get("iron_mq", None)
# SQS broker
- SQS = conf.get('sqs', None)
+ SQS = conf.get("sqs", None)
# ORM broker
- ORM = conf.get('orm', None)
+ ORM = conf.get("orm", None)
# Custom broker class
- BROKER_CLASS = conf.get('broker_class', None)
+ BROKER_CLASS = conf.get("broker_class", None)
# Database Poll
- POLL = conf.get('poll', 0.2)
+ POLL = conf.get("poll", 0.2)
# MongoDB broker
- MONGO = conf.get('mongo', None)
- MONGO_DB = conf.get('mongo_db', None)
+ MONGO = conf.get("mongo", None)
+ MONGO_DB = conf.get("mongo_db", None)
# Name of the cluster or site. For when you run multiple sites on one redis server
- PREFIX = conf.get('name', 'default')
+ PREFIX = conf.get("name", "default")
# Log output level
- LOG_LEVEL = conf.get('log_level', 'INFO')
+ LOG_LEVEL = conf.get("log_level", "INFO")
# Maximum number of successful tasks kept in the database. 0 saves everything. -1 saves none
# Failures are always saved
- SAVE_LIMIT = conf.get('save_limit', 250)
+ SAVE_LIMIT = conf.get("save_limit", 250)
# Guard loop sleep in seconds. Should be between 0 and 60 seconds.
- GUARD_CYCLE = conf.get('guard_cycle', 0.5)
+ GUARD_CYCLE = conf.get("guard_cycle", 0.5)
# Disable the scheduler
- SCHEDULER = conf.get('scheduler', True)
+ SCHEDULER = conf.get("scheduler", True)
# Number of workers in the pool. Default is cpu count if implemented, otherwise 4.
- WORKERS = conf.get('workers', False)
+ WORKERS = conf.get("workers", False)
if not WORKERS:
try:
WORKERS = cpu_count()
@@ -96,62 +98,62 @@ class Conf(object):
WORKERS = 4
# Option to undaemonize the workers and allow them to spawn child processes
- DAEMONIZE_WORKERS = conf.get('daemonize_workers', True)
+ DAEMONIZE_WORKERS = conf.get("daemonize_workers", True)
# Maximum number of tasks that each cluster can work on
- QUEUE_LIMIT = conf.get('queue_limit', int(WORKERS) ** 2)
+ QUEUE_LIMIT = conf.get("queue_limit", int(WORKERS) ** 2)
# Sets compression of redis packages
- COMPRESSED = conf.get('compress', False)
+ COMPRESSED = conf.get("compress", False)
# Number of tasks each worker can handle before it gets recycled. Useful for releasing memory
- RECYCLE = conf.get('recycle', 500)
+ RECYCLE = conf.get("recycle", 500)
# Number of seconds to wait for a worker to finish.
- TIMEOUT = conf.get('timeout', None)
+ TIMEOUT = conf.get("timeout", None)
# Whether to acknowledge unsuccessful tasks.
# This causes failed tasks to be considered delivered, thereby removing them from
# the task queue. Defaults to False.
- ACK_FAILURES = conf.get('ack_failures', False)
+ ACK_FAILURES = conf.get("ack_failures", False)
# Number of seconds to wait for acknowledgement before retrying a task
# Only works with brokers that guarantee delivery. Defaults to 60 seconds.
- RETRY = conf.get('retry', 60)
+ RETRY = conf.get("retry", 60)
# Sets the amount of tasks the cluster will try to pop off the broker.
# If it supports bulk gets.
- BULK = conf.get('bulk', 1)
+ BULK = conf.get("bulk", 1)
# The Django Admin label for this app
- LABEL = conf.get('label', 'Django Q')
+ LABEL = conf.get("label", "Django Q")
# Sets the number of processors for each worker, defaults to all.
- CPU_AFFINITY = conf.get('cpu_affinity', 0)
+ CPU_AFFINITY = conf.get("cpu_affinity", 0)
# Global sync option to for debugging
- SYNC = conf.get('sync', False)
+ SYNC = conf.get("sync", False)
# The Django cache to use
- CACHE = conf.get('cache', 'default')
+ CACHE = conf.get("cache", "default")
# Use the cache as result backend. Can be 'True' or an integer representing the global cache timeout.
# i.e 'cached: 60' , will make all results go the cache and expire in 60 seconds.
- CACHED = conf.get('cached', False)
+ CACHED = conf.get("cached", False)
# If set to False the scheduler won't execute tasks in the past.
# Instead it will run once and reschedule the next run in the future. Defaults to True.
- CATCH_UP = conf.get('catch_up', True)
+ CATCH_UP = conf.get("catch_up", True)
# Use the secret key for package signing
# Django itself should raise an error if it's not configured
SECRET_KEY = settings.SECRET_KEY
# The redis stats key
- Q_STAT = 'django_q:{}:cluster'.format(PREFIX)
+ Q_STAT = f"django_q:{PREFIX}:cluster"
# Optional error reporting setup
- ERROR_REPORTER = conf.get('error_reporter', {})
+ ERROR_REPORTER = conf.get("error_reporter", {})
# OSX doesn't implement qsize because of missing sem_getvalue()
try:
@@ -160,28 +162,33 @@ class Conf(object):
QSIZE = False
# Getting the signal names
- SIGNAL_NAMES = dict((getattr(signal, n), n) for n in dir(signal) if n.startswith('SIG') and '_' not in n)
+ SIGNAL_NAMES = dict(
+ (getattr(signal, n), n)
+ for n in dir(signal)
+ if n.startswith("SIG") and "_" not in n
+ )
# Translators: Cluster status descriptions
- STARTING = _('Starting')
- WORKING = _('Working')
+ STARTING = _("Starting")
+ WORKING = _("Working")
IDLE = _("Idle")
- STOPPED = _('Stopped')
- STOPPING = _('Stopping')
+ STOPPED = _("Stopped")
+ STOPPING = _("Stopping")
# to manage workarounds during testing
- TESTING = conf.get('testing', False)
+ TESTING = conf.get("testing", False)
# logger
-logger = logging.getLogger('django-q')
+logger = logging.getLogger("django-q")
# Set up standard logging handler in case there is none
if not logger.handlers:
logger.setLevel(level=getattr(logging, Conf.LOG_LEVEL))
logger.propagate = False
- formatter = logging.Formatter(fmt='%(asctime)s [Q] %(levelname)s %(message)s',
- datefmt='%H:%M:%S')
+ formatter = logging.Formatter(
+ fmt="%(asctime)s [Q] %(levelname)s %(message)s", datefmt="%H:%M:%S"
+ )
handler = logging.StreamHandler()
handler.setFormatter(formatter)
logger.addHandler(handler)
@@ -209,7 +216,8 @@ if Conf.ERROR_REPORTER:
# and instantiate an ErrorReporter using the provided config
for name, conf in error_conf.items():
for entry in pkg_resources.iter_entry_points(
- 'djangoq.errorreporters', name):
+ "djangoq.errorreporters", name
+ ):
Reporter = entry.load()
reporters.append(Reporter(**conf))
error_reporter = ErrorReporter(reporters)
@@ -221,9 +229,11 @@ else:
# get parent pid compatibility
def get_ppid():
- if hasattr(os, 'getppid'):
+ if hasattr(os, "getppid"):
return os.getppid()
elif psutil:
return psutil.Process(os.getpid()).ppid()
else:
- raise OSError('Your OS does not support `os.getppid`. Please install `psutil` as an alternative provider.')
+ raise OSError(
+ "Your OS does not support `os.getppid`. Please install `psutil` as an alternative provider."
+ )
diff --git a/django_q/models.py b/django_q/models.py
index e501555..86549ed 100644
--- a/django_q/models.py
+++ b/django_q/models.py
@@ -35,9 +35,15 @@ class Task(models.Model):
@staticmethod
def get_result_group(group_id, failures=False):
if failures:
- values = Task.objects.filter(group=group_id).values_list('result', flat=True)
+ values = Task.objects.filter(group=group_id).values_list(
+ "result", flat=True
+ )
else:
- values = Task.objects.filter(group=group_id).exclude(success=False).values_list('result', flat=True)
+ values = (
+ Task.objects.filter(group=group_id)
+ .exclude(success=False)
+ .values_list("result", flat=True)
+ )
return decode_results(values)
def group_result(self, failures=False):
@@ -86,76 +92,88 @@ class Task(models.Model):
return truncatechars(self.result, 100)
def __unicode__(self):
- return u'{}'.format(self.name or self.id)
+ return f"{self.name or self.id}"
class Meta:
- app_label = 'django_q'
- ordering = ['-stopped']
+ app_label = "django_q"
+ ordering = ["-stopped"]
class SuccessManager(models.Manager):
def get_queryset(self):
- return super(SuccessManager, self).get_queryset().filter(
- success=True)
+ return super(SuccessManager, self).get_queryset().filter(success=True)
class Success(Task):
objects = SuccessManager()
class Meta:
- app_label = 'django_q'
- verbose_name = _('Successful task')
- verbose_name_plural = _('Successful tasks')
- ordering = ['-stopped']
+ app_label = "django_q"
+ verbose_name = _("Successful task")
+ verbose_name_plural = _("Successful tasks")
+ ordering = ["-stopped"]
proxy = True
class FailureManager(models.Manager):
def get_queryset(self):
- return super(FailureManager, self).get_queryset().filter(
- success=False)
+ return super(FailureManager, self).get_queryset().filter(success=False)
class Failure(Task):
objects = FailureManager()
class Meta:
- app_label = 'django_q'
- verbose_name = _('Failed task')
- verbose_name_plural = _('Failed tasks')
- ordering = ['-stopped']
+ app_label = "django_q"
+ verbose_name = _("Failed task")
+ verbose_name_plural = _("Failed tasks")
+ ordering = ["-stopped"]
proxy = True
class Schedule(models.Model):
name = models.CharField(max_length=100, null=True, blank=True)
- func = models.CharField(max_length=256, help_text='e.g. module.tasks.function')
- hook = models.CharField(max_length=256, null=True, blank=True, help_text='e.g. module.tasks.result_function')
- args = models.TextField(null=True, blank=True, help_text=_("e.g. 1, 2, 'John'"))
- kwargs = models.TextField(null=True, blank=True, help_text=_("e.g. x=1, y=2, name='John'"))
- ONCE = 'O'
- MINUTES = 'I'
- HOURLY = 'H'
- DAILY = 'D'
- WEEKLY = 'W'
- MONTHLY = 'M'
- QUARTERLY = 'Q'
- YEARLY = 'Y'
- TYPE = (
- (ONCE, _('Once')),
- (MINUTES, _('Minutes')),
- (HOURLY, _('Hourly')),
- (DAILY, _('Daily')),
- (WEEKLY, _('Weekly')),
- (MONTHLY, _('Monthly')),
- (QUARTERLY, _('Quarterly')),
- (YEARLY, _('Yearly')),
+ func = models.CharField(max_length=256, help_text="e.g. module.tasks.function")
+ hook = models.CharField(
+ max_length=256,
+ null=True,
+ blank=True,
+ help_text="e.g. module.tasks.result_function",
+ )
+ args = models.TextField(null=True, blank=True, help_text=_("e.g. 1, 2, 'John'"))
+ kwargs = models.TextField(
+ null=True, blank=True, help_text=_("e.g. x=1, y=2, name='John'")
+ )
+ ONCE = "O"
+ MINUTES = "I"
+ HOURLY = "H"
+ DAILY = "D"
+ WEEKLY = "W"
+ MONTHLY = "M"
+ QUARTERLY = "Q"
+ YEARLY = "Y"
+ TYPE = (
+ (ONCE, _("Once")),
+ (MINUTES, _("Minutes")),
+ (HOURLY, _("Hourly")),
+ (DAILY, _("Daily")),
+ (WEEKLY, _("Weekly")),
+ (MONTHLY, _("Monthly")),
+ (QUARTERLY, _("Quarterly")),
+ (YEARLY, _("Yearly")),
+ )
+ schedule_type = models.CharField(
+ max_length=1, choices=TYPE, default=TYPE[0][0], verbose_name=_("Schedule Type")
+ )
+ minutes = models.PositiveSmallIntegerField(
+ null=True, blank=True, help_text=_("Number of minutes for the Minutes type")
+ )
+ repeats = models.IntegerField(
+ default=-1, verbose_name=_("Repeats"), help_text=_("n = n times, -1 = forever")
+ )
+ next_run = models.DateTimeField(
+ verbose_name=_("Next Run"), default=timezone.now, null=True
)
- schedule_type = models.CharField(max_length=1, choices=TYPE, default=TYPE[0][0], verbose_name=_('Schedule Type'))
- minutes = models.PositiveSmallIntegerField(null=True, blank=True,
- help_text=_('Number of minutes for the Minutes type'))
- repeats = models.IntegerField(default=-1, verbose_name=_('Repeats'), help_text=_('n = n times, -1 = forever'))
- next_run = models.DateTimeField(verbose_name=_('Next Run'), default=timezone.now, null=True)
task = models.CharField(max_length=100, null=True, editable=False)
def success(self):
@@ -166,10 +184,10 @@ class Schedule(models.Model):
if self.task and Task.objects.filter(id=self.task):
task = Task.objects.get(id=self.task)
if task.success:
- url = reverse('admin:django_q_success_change', args=(task.id,))
+ url = reverse("admin:django_q_success_change", args=(task.id,))
else:
- url = reverse('admin:django_q_failure_change', args=(task.id,))
- return format_html('[{}]'.format(url, task.name))
+ url = reverse("admin:django_q_failure_change", args=(task.id,))
+ return format_html(f'[{task.name}]')
return None
def __unicode__(self):
@@ -179,10 +197,10 @@ class Schedule(models.Model):
last_run.allow_tags = True
class Meta:
- app_label = 'django_q'
- verbose_name = _('Scheduled task')
- verbose_name_plural = _('Scheduled tasks')
- ordering = ['next_run']
+ app_label = "django_q"
+ verbose_name = _("Scheduled task")
+ verbose_name_plural = _("Scheduled tasks")
+ ordering = ["next_run"]
class OrmQ(models.Model):
@@ -194,23 +212,23 @@ class OrmQ(models.Model):
return SignedPackage.loads(self.payload)
def func(self):
- return self.task()['func']
+ return self.task()["func"]
def task_id(self):
- return self.task()['id']
+ return self.task()["id"]
def name(self):
- return self.task()['name']
+ return self.task()["name"]
class Meta:
- app_label = 'django_q'
- verbose_name = _('Queued task')
- verbose_name_plural = _('Queued tasks')
+ app_label = "django_q"
+ verbose_name = _("Queued task")
+ verbose_name_plural = _("Queued tasks")
# Backwards compatibility for Django 1.7
def decode_results(values):
- if get_version().split('.')[1] == '7':
+ if get_version().split(".")[1] == "7":
# decode values in 1.7
return [dbsafe_decode(v) for v in values]
return values
diff --git a/django_q/monitor.py b/django_q/monitor.py
index bccb3ac..488eda8 100644
--- a/django_q/monitor.py
+++ b/django_q/monitor.py
@@ -24,20 +24,44 @@ def monitor(run_once=False, broker=None):
with term.fullscreen(), term.hidden_cursor(), term.cbreak():
val = None
start_width = int(term.width / 8)
- while val not in (u'q', u'Q',):
+ while val not in ("q", "Q",):
col_width = int(term.width / 8)
# In case of resize
if col_width != start_width:
print(term.clear())
start_width = col_width
- print(term.move(0, 0) + term.black_on_green(term.center(_('Host'), width=col_width - 1)))
- print(term.move(0, 1 * col_width) + term.black_on_green(term.center(_('Id'), width=col_width - 1)))
- print(term.move(0, 2 * col_width) + term.black_on_green(term.center(_('State'), width=col_width - 1)))
- print(term.move(0, 3 * col_width) + term.black_on_green(term.center(_('Pool'), width=col_width - 1)))
- print(term.move(0, 4 * col_width) + term.black_on_green(term.center(_('TQ'), width=col_width - 1)))
- print(term.move(0, 5 * col_width) + term.black_on_green(term.center(_('RQ'), width=col_width - 1)))
- print(term.move(0, 6 * col_width) + term.black_on_green(term.center(_('RC'), width=col_width - 1)))
- print(term.move(0, 7 * col_width) + term.black_on_green(term.center(_('Up'), width=col_width - 1)))
+ print(
+ term.move(0, 0)
+ + term.black_on_green(term.center(_("Host"), width=col_width - 1))
+ )
+ print(
+ term.move(0, 1 * col_width)
+ + term.black_on_green(term.center(_("Id"), width=col_width - 1))
+ )
+ print(
+ term.move(0, 2 * col_width)
+ + term.black_on_green(term.center(_("State"), width=col_width - 1))
+ )
+ print(
+ term.move(0, 3 * col_width)
+ + term.black_on_green(term.center(_("Pool"), width=col_width - 1))
+ )
+ print(
+ term.move(0, 4 * col_width)
+ + term.black_on_green(term.center(_("TQ"), width=col_width - 1))
+ )
+ print(
+ term.move(0, 5 * col_width)
+ + term.black_on_green(term.center(_("RQ"), width=col_width - 1))
+ )
+ print(
+ term.move(0, 6 * col_width)
+ + term.black_on_green(term.center(_("RC"), width=col_width - 1))
+ )
+ print(
+ term.move(0, 7 * col_width)
+ + term.black_on_green(term.center(_("Up"), width=col_width - 1))
+ )
i = 2
stats = Stat.get_all(broker=broker)
print(term.clear_eos())
@@ -69,36 +93,83 @@ def monitor(run_once=False, broker=None):
uptime = (timezone.now() - stat.tob).total_seconds()
hours, remainder = divmod(uptime, 3600)
minutes, seconds = divmod(remainder, 60)
- uptime = '%d:%02d:%02d' % (hours, minutes, seconds)
+ uptime = "%d:%02d:%02d" % (hours, minutes, seconds)
# print to the terminal
- print(term.move(i, 0) + term.center(stat.host[:col_width - 1], width=col_width - 1))
- print(term.move(i, 1 * col_width) + term.center(str(stat.cluster_id)[-8:], width=col_width - 1))
- print(term.move(i, 2 * col_width) + term.center(status, width=col_width - 1))
- print(term.move(i, 3 * col_width) + term.center(workers, width=col_width - 1))
- print(term.move(i, 4 * col_width) + term.center(tasks, width=col_width - 1))
- print(term.move(i, 5 * col_width) + term.center(results, width=col_width - 1))
- print(term.move(i, 6 * col_width) + term.center(stat.reincarnations, width=col_width - 1))
- print(term.move(i, 7 * col_width) + term.center(uptime, width=col_width - 1))
+ print(
+ term.move(i, 0)
+ + term.center(stat.host[: col_width - 1], width=col_width - 1)
+ )
+ print(
+ term.move(i, 1 * col_width)
+ + term.center(str(stat.cluster_id)[-8:], width=col_width - 1)
+ )
+ print(
+ term.move(i, 2 * col_width)
+ + term.center(status, width=col_width - 1)
+ )
+ print(
+ term.move(i, 3 * col_width)
+ + term.center(workers, width=col_width - 1)
+ )
+ print(
+ term.move(i, 4 * col_width)
+ + term.center(tasks, width=col_width - 1)
+ )
+ print(
+ term.move(i, 5 * col_width)
+ + term.center(results, width=col_width - 1)
+ )
+ print(
+ term.move(i, 6 * col_width)
+ + term.center(stat.reincarnations, width=col_width - 1)
+ )
+ print(
+ term.move(i, 7 * col_width)
+ + term.center(uptime, width=col_width - 1)
+ )
i += 1
# bottom bar
i += 1
queue_size = broker.queue_size()
lock_size = broker.lock_size()
if lock_size:
- queue_size = '{}({})'.format(queue_size, lock_size)
- print(term.move(i, 0) + term.white_on_cyan(term.center(broker.info(), width=col_width * 2)))
- print(term.move(i, 2 * col_width) + term.black_on_cyan(term.center(_('Queued'), width=col_width)))
- print(term.move(i, 3 * col_width) + term.white_on_cyan(term.center(queue_size, width=col_width)))
- print(term.move(i, 4 * col_width) + term.black_on_cyan(term.center(_('Success'), width=col_width)))
- print(term.move(i, 5 * col_width) + term.white_on_cyan(
- term.center(models.Success.objects.count(), width=col_width)))
- print(term.move(i, 6 * col_width) + term.black_on_cyan(term.center(_('Failures'), width=col_width)))
- print(term.move(i, 7 * col_width) + term.white_on_cyan(
- term.center(models.Failure.objects.count(), width=col_width)))
+ queue_size = f"{queue_size}({lock_size})"
+ print(
+ term.move(i, 0)
+ + term.white_on_cyan(term.center(broker.info(), width=col_width * 2))
+ )
+ print(
+ term.move(i, 2 * col_width)
+ + term.black_on_cyan(term.center(_("Queued"), width=col_width))
+ )
+ print(
+ term.move(i, 3 * col_width)
+ + term.white_on_cyan(term.center(queue_size, width=col_width))
+ )
+ print(
+ term.move(i, 4 * col_width)
+ + term.black_on_cyan(term.center(_("Success"), width=col_width))
+ )
+ print(
+ term.move(i, 5 * col_width)
+ + term.white_on_cyan(
+ term.center(models.Success.objects.count(), width=col_width)
+ )
+ )
+ print(
+ term.move(i, 6 * col_width)
+ + term.black_on_cyan(term.center(_("Failures"), width=col_width))
+ )
+ print(
+ term.move(i, 7 * col_width)
+ + term.white_on_cyan(
+ term.center(models.Failure.objects.count(), width=col_width)
+ )
+ )
# for testing
if run_once:
return Stat.get_all(broker=broker)
- print(term.move(i + 2, 0) + term.center(_('[Press q to quit]')))
+ print(term.move(i + 2, 0) + term.center(_("[Press q to quit]")))
val = term.inkey(timeout=1)
@@ -117,15 +188,19 @@ def info(broker=None):
reincarnations += cluster.reincarnations
# calculate tasks pm and avg exec time
tasks_per = 0
- per = _('day')
+ per = _("day")
exec_time = 0
- last_tasks = models.Success.objects.filter(stopped__gte=timezone.now() - timedelta(hours=24))
+ last_tasks = models.Success.objects.filter(
+ stopped__gte=timezone.now() - timedelta(hours=24)
+ )
tasks_per_day = last_tasks.count()
if tasks_per_day > 0:
# average execution time over the last 24 hours
- if not connection.vendor == 'sqlite':
- exec_time = last_tasks.aggregate(time_taken=Sum(F('stopped') - F('started')))
- exec_time = exec_time['time_taken'].total_seconds() / tasks_per_day
+ if not connection.vendor == "sqlite":
+ exec_time = last_tasks.aggregate(
+ time_taken=Sum(F("stopped") - F("started"))
+ )
+ exec_time = exec_time["time_taken"].total_seconds() / tasks_per_day
else:
# can't sum timedeltas on sqlite
for t in last_tasks:
@@ -134,58 +209,66 @@ def info(broker=None):
# tasks per second/minute/hour/day in the last 24 hours
if tasks_per_day > 24 * 60 * 60:
tasks_per = tasks_per_day / (24 * 60 * 60)
- per = _('second')
+ per = _("second")
elif tasks_per_day > 24 * 60:
tasks_per = tasks_per_day / (24 * 60)
- per = _('minute')
+ per = _("minute")
elif tasks_per_day > 24:
tasks_per = tasks_per_day / 24
- per = _('hour')
+ per = _("hour")
else:
tasks_per = tasks_per_day
# print to terminal
print(term.clear_eos())
col_width = int(term.width / 6)
- print(term.black_on_green(
- term.center(
- _('-- {} {} on {} --').format(Conf.PREFIX.capitalize(), '.'.join(str(v) for v in VERSION),
- broker.info()))))
- print(term.cyan(_('Clusters')) +
- term.move_x(1 * col_width) +
- term.white(str(clusters)) +
- term.move_x(2 * col_width) +
- term.cyan(_('Workers')) +
- term.move_x(3 * col_width) +
- term.white(str(workers)) +
- term.move_x(4 * col_width) +
- term.cyan(_('Restarts')) +
- term.move_x(5 * col_width) +
- term.white(str(reincarnations))
- )
- print(term.cyan(_('Queued')) +
- term.move_x(1 * col_width) +
- term.white(str(broker.queue_size())) +
- term.move_x(2 * col_width) +
- term.cyan(_('Successes')) +
- term.move_x(3 * col_width) +
- term.white(str(models.Success.objects.count())) +
- term.move_x(4 * col_width) +
- term.cyan(_('Failures')) +
- term.move_x(5 * col_width) +
- term.white(str(models.Failure.objects.count()))
- )
- print(term.cyan(_('Schedules')) +
- term.move_x(1 * col_width) +
- term.white(str(models.Schedule.objects.count())) +
- term.move_x(2 * col_width) +
- term.cyan(_('Tasks/{}'.format(per))) +
- term.move_x(3 * col_width) +
- term.white('{0:.2f}'.format(tasks_per)) +
- term.move_x(4 * col_width) +
- term.cyan(_('Avg time')) +
- term.move_x(5 * col_width) +
- term.white('{0:.4f}'.format(exec_time))
- )
+ print(
+ term.black_on_green(
+ term.center(
+ _(
+ f'-- {Conf.PREFIX.capitalize()} { ".".join(str(v) for v in VERSION)} on {broker.info()} --'
+ )
+ )
+ )
+ )
+ print(
+ term.cyan(_("Clusters"))
+ + term.move_x(1 * col_width)
+ + term.white(str(clusters))
+ + term.move_x(2 * col_width)
+ + term.cyan(_("Workers"))
+ + term.move_x(3 * col_width)
+ + term.white(str(workers))
+ + term.move_x(4 * col_width)
+ + term.cyan(_("Restarts"))
+ + term.move_x(5 * col_width)
+ + term.white(str(reincarnations))
+ )
+ print(
+ term.cyan(_("Queued"))
+ + term.move_x(1 * col_width)
+ + term.white(str(broker.queue_size()))
+ + term.move_x(2 * col_width)
+ + term.cyan(_("Successes"))
+ + term.move_x(3 * col_width)
+ + term.white(str(models.Success.objects.count()))
+ + term.move_x(4 * col_width)
+ + term.cyan(_("Failures"))
+ + term.move_x(5 * col_width)
+ + term.white(str(models.Failure.objects.count()))
+ )
+ print(
+ term.cyan(_("Schedules"))
+ + term.move_x(1 * col_width)
+ + term.white(str(models.Schedule.objects.count()))
+ + term.move_x(2 * col_width)
+ + term.cyan(_(f"Tasks/{per}"))
+ + term.move_x(3 * col_width)
+ + term.white(f"{tasks_per:.2f}")
+ + term.move_x(4 * col_width)
+ + term.cyan(_("Avg time"))
+ + term.move_x(5 * col_width)
+ + term.white(f"{exec_time:.4f}")
+ )
return True
@@ -196,5 +279,5 @@ def get_ids():
for s in stat:
print(s.cluster_id)
else:
- print('No clusters appear to be running.')
+ print("No clusters appear to be running.")
return True
diff --git a/django_q/signals.py b/django_q/signals.py
index f12436e..0205d5e 100644
--- a/django_q/signals.py
+++ b/django_q/signals.py
@@ -14,16 +14,22 @@ def call_hook(sender, instance, **kwargs):
f = instance.hook
if not callable(f):
try:
- module, func = f.rsplit('.', 1)
+ module, func = f.rsplit(".", 1)
m = importlib.import_module(module)
f = getattr(m, func)
except (ValueError, ImportError, AttributeError):
- logger.error(_('malformed return hook \'{}\' for [{}]').format(instance.hook, instance.name))
+ logger.error(
+ _(f"malformed return hook '{instance.hook}' for [{instance.name}]")
+ )
return
try:
f(instance)
except Exception as e:
- logger.error(_('return hook {} failed on [{}] because {}').format(instance.hook, instance.name, e))
+ logger.error(
+ _(
+ f"return hook {instance.hook} failed on [{instance.name}] because {str(e)}"
+ )
+ )
pre_enqueue = Signal(providing_args=["task"])
diff --git a/django_q/status.py b/django_q/status.py
index 06402a9..f3f9450 100644
--- a/django_q/status.py
+++ b/django_q/status.py
@@ -28,7 +28,9 @@ class Stat(Status):
"""Status object for Cluster monitoring."""
def __init__(self, sentinel):
- super(Stat, self).__init__(sentinel.parent_pid or sentinel.pid, cluster_id=sentinel.cluster_id)
+ super(Stat, self).__init__(
+ sentinel.parent_pid or sentinel.pid, cluster_id=sentinel.cluster_id
+ )
self.broker = sentinel.broker or get_broker()
self.tob = sentinel.tob
self.reincarnations = sentinel.reincarnations
@@ -61,7 +63,7 @@ class Stat(Status):
:param cluster_id: cluster ID
:return: redis key for the cluster statistic
"""
- return '{}:{}'.format(Conf.Q_STAT, cluster_id)
+ return f"{Conf.Q_STAT}:{cluster_id}"
def save(self):
try:
@@ -99,7 +101,7 @@ class Stat(Status):
if not broker:
broker = get_broker()
stats = []
- packs = broker.get_stats('{}:*'.format(Conf.Q_STAT)) or []
+ packs = broker.get_stats(f"{Conf.Q_STAT}:*") or []
for pack in packs:
try:
stats.append(SignedPackage.loads(pack))
@@ -110,5 +112,5 @@ class Stat(Status):
def __getstate__(self):
# Don't pickle the redis connection
state = dict(self.__dict__)
- del state['broker']
+ del state["broker"]
return state
diff --git a/django_q/tasks.py b/django_q/tasks.py
index ea2c9b4..ccc333b 100644
--- a/django_q/tasks.py
+++ b/django_q/tasks.py
@@ -8,6 +8,7 @@ from django.utils import timezone
from multiprocessing import Value
from django_q.brokers import get_broker
+
# local
from django_q.conf import Conf, logger
from django_q.humanhash import uuid
@@ -21,15 +22,30 @@ def async_task(func, *args, **kwargs):
"""Queue a task for the cluster."""
keywords = kwargs.copy()
opt_keys = (
- 'hook', 'group', 'save', 'sync', 'cached', 'ack_failure', 'iter_count', 'iter_cached', 'chain', 'broker', 'timeout')
- q_options = keywords.pop('q_options', {})
+ "hook",
+ "group",
+ "save",
+ "sync",
+ "cached",
+ "ack_failure",
+ "iter_count",
+ "iter_cached",
+ "chain",
+ "broker",
+ "timeout",
+ )
+ q_options = keywords.pop("q_options", {})
# get an id
tag = uuid()
# build the task package
- task = {'id': tag[1],
- 'name': keywords.pop('task_name', None) or q_options.pop('task_name', None) or tag[0],
- 'func': func,
- 'args': args}
+ task = {
+ "id": tag[1],
+ "name": keywords.pop("task_name", None)
+ or q_options.pop("task_name", None)
+ or tag[0],
+ "func": func,
+ "args": args,
+ }
# push optionals
for key in opt_keys:
if q_options and key in q_options:
@@ -37,28 +53,28 @@ def async_task(func, *args, **kwargs):
elif key in keywords:
task[key] = keywords.pop(key)
# don't serialize the broker
- broker = task.pop('broker', get_broker())
+ broker = task.pop("broker", get_broker())
# overrides
- if 'cached' not in task and Conf.CACHED:
- task['cached'] = Conf.CACHED
- if 'sync' not in task and Conf.SYNC:
- task['sync'] = Conf.SYNC
- if 'ack_failure' not in task and Conf.ACK_FAILURES:
- task['ack_failure'] = Conf.ACK_FAILURES
+ if "cached" not in task and Conf.CACHED:
+ task["cached"] = Conf.CACHED
+ if "sync" not in task and Conf.SYNC:
+ task["sync"] = Conf.SYNC
+ if "ack_failure" not in task and Conf.ACK_FAILURES:
+ task["ack_failure"] = Conf.ACK_FAILURES
# finalize
- task['kwargs'] = keywords
- task['started'] = timezone.now()
+ task["kwargs"] = keywords
+ task["started"] = timezone.now()
# signal it
pre_enqueue.send(sender="django_q", task=task)
# sign it
pack = SignedPackage.dumps(task)
- if task.get('sync', False):
+ if task.get("sync", False):
return _sync(pack)
# push it
enqueue_id = broker.enqueue(pack)
- logger.info('Enqueued {}'.format(enqueue_id))
- logger.debug('Pushed {}'.format(tag))
- return task['id']
+ logger.info(f"Enqueued {enqueue_id}")
+ logger.debug(f"Pushed {tag}")
+ return task["id"]
def schedule(func, *args, **kwargs):
@@ -77,28 +93,29 @@ def schedule(func, *args, **kwargs):
:return: the schedule object.
:rtype: Schedule
"""
- name = kwargs.pop('name', None)
- hook = kwargs.pop('hook', None)
- schedule_type = kwargs.pop('schedule_type', Schedule.ONCE)
- minutes = kwargs.pop('minutes', None)
- repeats = kwargs.pop('repeats', -1)
- next_run = kwargs.pop('next_run', timezone.now())
+ name = kwargs.pop("name", None)
+ hook = kwargs.pop("hook", None)
+ schedule_type = kwargs.pop("schedule_type", Schedule.ONCE)
+ minutes = kwargs.pop("minutes", None)
+ repeats = kwargs.pop("repeats", -1)
+ next_run = kwargs.pop("next_run", timezone.now())
# check for name duplicates instead of am unique constraint
if name and Schedule.objects.filter(name=name).exists():
raise IntegrityError("A schedule with the same name already exists.")
# create and return the schedule
- return Schedule.objects.create(name=name,
- func=func,
- hook=hook,
- args=args,
- kwargs=kwargs,
- schedule_type=schedule_type,
- minutes=minutes,
- repeats=repeats,
- next_run=next_run
- )
+ return Schedule.objects.create(
+ name=name,
+ func=func,
+ hook=hook,
+ args=args,
+ kwargs=kwargs,
+ schedule_type=schedule_type,
+ minutes=minutes,
+ repeats=repeats,
+ next_run=next_run,
+ )
def result(task_id, wait=0, cached=Conf.CACHED):
@@ -133,9 +150,9 @@ def result_cached(task_id, wait=0, broker=None):
broker = get_broker()
start = time()
while True:
- r = broker.cache.get('{}:{}'.format(broker.list_key, task_id))
+ r = broker.cache.get(f"{broker.list_key}:{task_id}")
if r:
- return SignedPackage.loads(r)['result']
+ return SignedPackage.loads(r)["result"]
if (time() - start) * 1000 >= wait >= 0:
break
sleep(0.01)
@@ -156,7 +173,11 @@ def result_group(group_id, failures=False, wait=0, count=None, cached=Conf.CACHE
start = time()
if count:
while True:
- if count_group(group_id) == count or wait and (time() - start) * 1000 >= wait >= 0:
+ if (
+ count_group(group_id) == count
+ or wait
+ and (time() - start) * 1000 >= wait >= 0
+ ):
break
sleep(0.01)
while True:
@@ -177,17 +198,21 @@ def result_group_cached(group_id, failures=False, wait=0, count=None, broker=Non
start = time()
if count:
while True:
- if count_group_cached(group_id) == count or wait and (time() - start) * 1000 >= wait > 0:
+ if (
+ count_group_cached(group_id) == count
+ or wait
+ and (time() - start) * 1000 >= wait > 0
+ ):
break
sleep(0.01)
while True:
- group_list = broker.cache.get('{}:{}:keys'.format(broker.list_key, group_id))
+ group_list = broker.cache.get(f"{broker.list_key}:{group_id}:keys")
if group_list:
result_list = []
for task_key in group_list:
task = SignedPackage.loads(broker.cache.get(task_key))
- if task['success'] or failures:
- result_list.append(task['result'])
+ if task["success"] or failures:
+ result_list.append(task["result"])
return result_list
if (time() - start) * 1000 >= wait >= 0:
break
@@ -226,19 +251,21 @@ def fetch_cached(task_id, wait=0, broker=None):
broker = get_broker()
start = time()
while True:
- r = broker.cache.get('{}:{}'.format(broker.list_key, task_id))
+ r = broker.cache.get(f"{broker.list_key}:{task_id}")
if r:
task = SignedPackage.loads(r)
- t = Task(id=task['id'],
- name=task['name'],
- func=task['func'],
- hook=task.get('hook'),
- args=task['args'],
- kwargs=task['kwargs'],
- started=task['started'],
- stopped=task['stopped'],
- result=task['result'],
- success=task['success'])
+ t = Task(
+ id=task["id"],
+ name=task["name"],
+ func=task["func"],
+ hook=task.get("hook"),
+ args=task["args"],
+ kwargs=task["kwargs"],
+ started=task["started"],
+ stopped=task["stopped"],
+ result=task["result"],
+ success=task["success"],
+ )
return t
if (time() - start) * 1000 >= wait >= 0:
break
@@ -259,7 +286,11 @@ def fetch_group(group_id, failures=True, wait=0, count=None, cached=Conf.CACHED)
start = time()
if count:
while True:
- if count_group(group_id) == count or wait and (time() - start) * 1000 >= wait >= 0:
+ if (
+ count_group(group_id) == count
+ or wait
+ and (time() - start) * 1000 >= wait >= 0
+ ):
break
sleep(0.01)
while True:
@@ -280,27 +311,33 @@ def fetch_group_cached(group_id, failures=True, wait=0, count=None, broker=None)
start = time()
if count:
while True:
- if count_group_cached(group_id) == count or wait and (time() - start) * 1000 >= wait >= 0:
+ if (
+ count_group_cached(group_id) == count
+ or wait
+ and (time() - start) * 1000 >= wait >= 0
+ ):
break
sleep(0.01)
while True:
- group_list = broker.cache.get('{}:{}:keys'.format(broker.list_key, group_id))
+ group_list = broker.cache.get(f"{broker.list_key}:{group_id}:keys")
if group_list:
task_list = []
for task_key in group_list:
task = SignedPackage.loads(broker.cache.get(task_key))
- if task['success'] or failures:
- t = Task(id=task['id'],
- name=task['name'],
- func=task['func'],
- hook=task.get('hook'),
- args=task['args'],
- kwargs=task['kwargs'],
- started=task['started'],
- stopped=task['stopped'],
- result=task['result'],
- group=task.get('group'),
- success=task['success'])
+ if task["success"] or failures:
+ t = Task(
+ id=task["id"],
+ name=task["name"],
+ func=task["func"],
+ hook=task.get("hook"),
+ args=task["args"],
+ kwargs=task["kwargs"],
+ started=task["started"],
+ stopped=task["stopped"],
+ result=task["result"],
+ group=task.get("group"),
+ success=task["success"],
+ )
task_list.append(t)
return task_list
if (time() - start) * 1000 >= wait >= 0:
@@ -329,14 +366,14 @@ def count_group_cached(group_id, failures=False, broker=None):
"""
if not broker:
broker = get_broker()
- group_list = broker.cache.get('{}:{}:keys'.format(broker.list_key, group_id))
+ group_list = broker.cache.get(f"{broker.list_key}:{group_id}:keys")
if group_list:
if not failures:
return len(group_list)
failure_count = 0
for task_key in group_list:
task = SignedPackage.loads(broker.cache.get(task_key))
- if not task['success']:
+ if not task["success"]:
failure_count += 1
return failure_count
@@ -362,7 +399,7 @@ def delete_group_cached(group_id, broker=None):
"""
if not broker:
broker = get_broker()
- group_key = '{}:{}:keys'.format(broker.list_key, group_id)
+ group_key = f"{broker.list_key}:{group_id}:keys"
group_list = broker.cache.get(group_key)
broker.cache.delete_many(group_list)
broker.cache.delete(group_key)
@@ -374,7 +411,7 @@ def delete_cached(task_id, broker=None):
"""
if not broker:
broker = get_broker()
- return broker.cache.delete('{}:{}'.format(broker.list_key, task_id))
+ return broker.cache.delete(f"{broker.list_key}:{task_id}")
def queue_size(broker=None):
@@ -398,17 +435,19 @@ def async_iter(func, args_iter, **kwargs):
iter_count = len(args_iter)
iter_group = uuid()[1]
# clean up the kwargs
- options = kwargs.get('q_options', kwargs)
- options.pop('hook', None)
- options['broker'] = options.get('broker', get_broker())
- options['group'] = iter_group
- options['iter_count'] = iter_count
- if options.get('cached', None):
- options['iter_cached'] = options['cached']
- options['cached'] = True
+ options = kwargs.get("q_options", kwargs)
+ options.pop("hook", None)
+ options["broker"] = options.get("broker", get_broker())
+ options["group"] = iter_group
+ options["iter_count"] = iter_count
+ if options.get("cached", None):
+ options["iter_cached"] = options["cached"]
+ options["cached"] = True
# save the original arguments
- broker = options['broker']
- broker.cache.set('{}:{}:args'.format(broker.list_key, iter_group), SignedPackage.dumps(args_iter))
+ broker = options["broker"]
+ broker.cache.set(
+ f"{broker.list_key}:{iter_group}:args", SignedPackage.dumps(args_iter)
+ )
for args in args_iter:
if not isinstance(args, tuple):
args = (args,)
@@ -432,11 +471,11 @@ def async_chain(chain, group=None, cached=Conf.CACHED, sync=Conf.SYNC, broker=No
args = task[1]
if len(task) > 2:
kwargs = task[2]
- kwargs['chain'] = chain
- kwargs['group'] = group
- kwargs['cached'] = cached
- kwargs['sync'] = sync
- kwargs['broker'] = broker or get_broker()
+ kwargs["chain"] = chain
+ kwargs["group"] = group
+ kwargs["cached"] = cached
+ kwargs["sync"] = sync
+ kwargs["broker"] = broker or get_broker()
async_task(task[0], *args, **kwargs)
return group
@@ -446,11 +485,19 @@ class Iter(object):
An async task with iterable arguments
"""
- def __init__(self, func=None, args=None, kwargs=None, cached=Conf.CACHED, sync=Conf.SYNC, broker=None):
+ def __init__(
+ self,
+ func=None,
+ args=None,
+ kwargs=None,
+ cached=Conf.CACHED,
+ sync=Conf.SYNC,
+ broker=None,
+ ):
self.func = func
self.args = args or []
self.kwargs = kwargs or {}
- self.id = ''
+ self.id = ""
self.broker = broker or get_broker()
self.cached = cached
self.sync = sync
@@ -470,9 +517,9 @@ class Iter(object):
Start queueing the tasks to the worker cluster
:return: the task id
"""
- self.kwargs['cached'] = self.cached
- self.kwargs['sync'] = self.sync
- self.kwargs['broker'] = self.broker
+ self.kwargs["cached"] = self.cached
+ self.kwargs["sync"] = self.sync
+ self.kwargs["broker"] = self.broker
self.id = async_iter(self.func, self.args, **self.kwargs)
self.started = True
return self.id
@@ -510,7 +557,7 @@ class Chain(object):
def __init__(self, chain=None, group=None, cached=Conf.CACHED, sync=Conf.SYNC):
self.chain = chain or []
- self.group = group or ''
+ self.group = group or ""
self.broker = get_broker()
self.cached = cached
self.sync = sync
@@ -533,8 +580,13 @@ class Chain(object):
Start queueing the chain to the worker cluster
:return: the chain's group id
"""
- self.group = async_chain(chain=self.chain[:], group=self.group, cached=self.cached, sync=self.sync,
- broker=self.broker)
+ self.group = async_chain(
+ chain=self.chain[:],
+ group=self.group,
+ cached=self.cached,
+ sync=self.sync,
+ broker=self.broker,
+ )
self.started = True
return self.group
@@ -545,7 +597,9 @@ class Chain(object):
:return: an unsorted list of results
"""
if self.started:
- return result_group(self.group, wait=wait, count=self.length(), cached=self.cached)
+ return result_group(
+ self.group, wait=wait, count=self.length(), cached=self.cached
+ )
def fetch(self, failures=True, wait=0):
"""
@@ -555,7 +609,13 @@ class Chain(object):
:return: an unsorted list of task objects
"""
if self.started:
- return fetch_group(self.group, failures=failures, wait=wait, count=self.length(), cached=self.cached)
+ return fetch_group(
+ self.group,
+ failures=failures,
+ wait=wait,
+ count=self.length(),
+ cached=self.cached,
+ )
def current(self):
"""
@@ -580,7 +640,7 @@ class AsyncTask(object):
"""
def __init__(self, func, *args, **kwargs):
- self.id = ''
+ self.id = ""
self.started = False
self.func = func
self.args = args
@@ -588,62 +648,62 @@ class AsyncTask(object):
@property
def broker(self):
- return self._get_option('broker', None)
+ return self._get_option("broker", None)
@broker.setter
def broker(self, value):
- self._set_option('broker', value)
+ self._set_option("broker", value)
@property
def sync(self):
- return self._get_option('sync', None)
+ return self._get_option("sync", None)
@sync.setter
def sync(self, value):
- self._set_option('sync', value)
+ self._set_option("sync", value)
@property
def save(self):
- return self._get_option('save', None)
+ return self._get_option("save", None)
@save.setter
def save(self, value):
- self._set_option('save', value)
+ self._set_option("save", value)
@property
def hook(self):
- return self._get_option('hook', None)
+ return self._get_option("hook", None)
@hook.setter
def hook(self, value):
- self._set_option('hook', value)
+ self._set_option("hook", value)
@property
def group(self):
- return self._get_option('group', None)
+ return self._get_option("group", None)
@group.setter
def group(self, value):
- self._set_option('group', value)
+ self._set_option("group", value)
@property
def cached(self):
- return self._get_option('cached', Conf.CACHED)
+ return self._get_option("cached", Conf.CACHED)
@cached.setter
def cached(self, value):
- self._set_option('cached', value)
+ self._set_option("cached", value)
def _set_option(self, key, value):
- if 'q_options' in self.kwargs:
- self.kwargs['q_options'][key] = value
+ if "q_options" in self.kwargs:
+ self.kwargs["q_options"][key] = value
else:
self.kwargs[key] = value
self.started = False
def _get_option(self, key, default=None):
- if 'q_options' in self.kwargs:
- return self.kwargs['q_options'].get(key, default)
+ if "q_options" in self.kwargs:
+ return self.kwargs["q_options"].get(key, default)
else:
return self.kwargs.get(key, default)
@@ -665,27 +725,40 @@ class AsyncTask(object):
def result_group(self, failures=False, wait=0, count=None):
if self.started and self.group:
- return result_group(self.group, failures=failures, wait=wait, count=count, cached=self.cached)
+ return result_group(
+ self.group,
+ failures=failures,
+ wait=wait,
+ count=count,
+ cached=self.cached,
+ )
def fetch_group(self, failures=True, wait=0, count=None):
if self.started and self.group:
- return fetch_group(self.group, failures=failures, wait=wait, count=count, cached=self.cached)
+ return fetch_group(
+ self.group,
+ failures=failures,
+ wait=wait,
+ count=count,
+ cached=self.cached,
+ )
def _sync(pack):
"""Simulate a package travelling through the cluster."""
from django_q.cluster import worker, monitor
+
task_queue = Queue()
result_queue = Queue()
task = SignedPackage.loads(pack)
task_queue.put(task)
- task_queue.put('STOP')
- worker(task_queue, result_queue, Value('f', -1))
- result_queue.put('STOP')
+ task_queue.put("STOP")
+ worker(task_queue, result_queue, Value("f", -1))
+ result_queue.put("STOP")
monitor(result_queue)
task_queue.close()
task_queue.join_thread()
result_queue.close()
result_queue.join_thread()
- return task['id']
+ return task["id"]