fixes connection creation on import

django_q/cluster.py

@@ -37,8 +37,8 @@ from django_q.brokers import get_broker
 
 
 class Cluster(object):
-    def __init__(self, broker=get_broker()):
+    def __init__(self, broker=None):
-        self.broker = broker
+        self.broker = broker or get_broker()
         self.sentinel = None
         self.stop_event = None
         self.start_event = None
@@ -106,14 +106,14 @@ class Cluster(object):
 
 
 class Sentinel(object):
-    def __init__(self, stop_event, start_event, broker=get_broker(), timeout=Conf.TIMEOUT, start=True):
+    def __init__(self, stop_event, start_event, 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)
         self.pid = current_process().pid
         self.parent_pid = get_ppid()
         self.name = current_process().name
-        self.broker = broker
+        self.broker = broker or get_broker()
         self.reincarnations = 0
         self.tob = timezone.now()
         self.stop_event = stop_event
@@ -287,12 +287,14 @@ class Sentinel(object):
         Stat(self).save()
 
 
-def pusher(task_queue, event, broker=get_broker()):
+def pusher(task_queue, event, broker=None):
     """
     Pulls tasks of the Redis List and puts them in the task queue
     :type task_queue: multiprocessing.Queue
     :type event: multiprocessing.Event
     """
+    if not broker:
+        broker = get_broker()
     logger.info(_('{} pushing tasks at {}').format(current_process().name, current_process().pid))
     while True:
         try:
@@ -318,11 +320,13 @@ def pusher(task_queue, event, broker=get_broker()):
     logger.info(_("{} stopped pushing tasks").format(current_process().name))
 
 
-def monitor(result_queue, broker=get_broker()):
+def monitor(result_queue, broker=None):
     """
     Gets finished tasks from the result queue and saves them to Django
     :type result_queue: multiprocessing.Queue
     """
+    if not broker:
+        broker = get_broker()
     name = current_process().name
     logger.info(_("{} monitoring at {}").format(name, current_process().pid))
     db.close_old_connections()
@@ -413,10 +417,12 @@ def save_task(task):
         logger.error(e)
 
 
-def scheduler(broker=get_broker()):
+def scheduler(broker=None):
     """
     Creates a task from a schedule at the scheduled time and schedules next run
     """
+    if not broker:
+        broker = get_broker()
     try:
         for s in Schedule.objects.exclude(repeats=0).filter(next_run__lt=timezone.now()):
             args = ()

django_q/monitor.py

@@ -10,13 +10,15 @@ from django.utils import timezone
 from django.utils.translation import ugettext as _
 
 # local
-from django_q.conf import Conf, redis_client
+from django_q.conf import Conf
 from django_q.status import Stat
 from django_q.brokers import get_broker
 from django_q import models
 
 
-def monitor(run_once=False, broker=get_broker()):
+def monitor(run_once=False, broker=None):
+    if not broker:
+        broker = get_broker()
     term = Terminal()
     broker.ping()
     with term.fullscreen(), term.hidden_cursor(), term.cbreak():
@@ -85,7 +87,9 @@ def monitor(run_once=False, broker=get_broker()):
             val = term.inkey(timeout=1)
 
 
-def info(broker=get_broker()):
+def info(broker=None):
+    if not broker:
+        broker = get_broker()
     term = Terminal()
     broker.ping()
     stat = Stat.get_all(broker=broker)

django_q/status.py

@@ -72,12 +72,14 @@ class Stat(Status):
         return self.done_q_size + self.task_q_size == 0
 
     @staticmethod
-    def get(cluster_id, broker=get_broker()):
+    def get(cluster_id, broker=None):
         """
         gets the current status for the cluster
         :param cluster_id: id of the cluster
         :return: Stat or Status
         """
+        if not broker:
+            broker = get_broker()
         pack = broker.get_stat(Stat.get_key(cluster_id))
         if pack:
             try:
@@ -87,12 +89,14 @@ class Stat(Status):
         return Status(cluster_id)
 
     @staticmethod
-    def get_all(broker=get_broker()):
+    def get_all(broker=None):
         """
         Get the status for all currently running clusters with the same prefix
         and secret key.
         :return: list of type Stat
         """
+        if not broker:
+            broker = get_broker()
         stats = []
         packs = broker.get_stats('{}:*'.format(Conf.Q_STAT)) or []
         for pack in packs:

django_q/tasks.py

@@ -152,16 +152,17 @@ def delete_group(group_id, tasks=False):
     return Task.delete_group(group_id, tasks)
 
 
-def queue_size(broker=get_broker()):
+def queue_size(broker=None):
     """
     Returns the current queue size.
     Note that this doesn't count any tasks currently being processed by workers.
 
-    :param list_key: optional list key
     :param broker: optional broker
     :return: current queue size
     :rtype: int
     """
+    if not broker:
+        broker = get_broker()
     return broker.queue_size()