fork/django-q2
1
0
Fork 0
mirror of https://github.com/django-q2/django-q2.git synced 2026-10-08 09:48:11 +08:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #297 from Eagllus/fix_199

Browse Source 
Changing the path location where Django-Q is inserted.
This commit is contained in:
Ilan Steemers
2018-03-13 15:22:14 +01:00
committed by GitHub
parent c657731cd0 430c177b92
commit 65afb63e19
4 changed files with 66 additions and 63 deletions
Download Patch File Download Diff File Expand all files Collapse all files
django_q/__init__.py
+4 -4
View File
@@ -1,9 +1,9 @@
import os # import os
import sys # import sys
import django import django
myPath = os.path.dirname(os.path.abspath(__file__)) # myPath = os.path.dirname(os.path.abspath(__file__))
sys.path.insert(0, myPath) # sys.path.insert(0, myPath)
VERSION = (0, 9, 2) VERSION = (0, 9, 2)
django_q/cluster.py
+7 -8
View File
@@ -21,12 +21,11 @@ from django.utils.translation import ugettext_lazy as _
from django import db from django import db
# Local # Local
import signing from django_q import tasks
import tasks
from django_q.compat import range from django_q.compat import range
from django_q.conf import Conf, logger, psutil, get_ppid, error_reporter, rollbar from django_q.conf import Conf, logger, psutil, get_ppid, error_reporter, rollbar
from django_q.models import Task, Success, Schedule from django_q.models import Task, Success, Schedule
from django_q.signing import SignedPackage, BadSignature
from django_q.status import Stat, Status from django_q.status import Stat, Status
from django_q.brokers import get_broker from django_q.brokers import get_broker
from django_q.signals import pre_execute from django_q.signals import pre_execute
@@ -297,8 +296,8 @@ def pusher(task_queue, event, broker=None):
ack_id = task[0] ack_id = task[0]
# unpack the task # unpack the task
try: try:
task = signing.SignedPackage.loads(task[1]) task = SignedPackage.loads(task[1])
except (TypeError, signing.BadSignature) as e: except (TypeError, BadSignature) as e:
logger.error(e) logger.error(e)
broker.fail(ack_id) broker.fail(ack_id)
continue continue
@@ -456,11 +455,11 @@ def save_cached(task, broker):
if iter_count and len(group_list) == iter_count - 1: if iter_count and len(group_list) == iter_count - 1:
group_args = '{}:{}:args'.format(broker.list_key, group) group_args = '{}:{}:args'.format(broker.list_key, group)
# collate the results into a Task result # collate the results into a Task result
results = [signing.SignedPackage.loads(broker.cache.get(k))['result'] for k in group_list] results = [SignedPackage.loads(broker.cache.get(k))['result'] for k in group_list]
results.append(task['result']) results.append(task['result'])
task['result'] = results task['result'] = results
task['id'] = group task['id'] = group
task['args'] = signing.SignedPackage.loads(broker.cache.get(group_args)) task['args'] = SignedPackage.loads(broker.cache.get(group_args))
task.pop('iter_count', None) task.pop('iter_count', None)
task.pop('group', None) task.pop('group', None)
if task.get('iter_cached', None): if task.get('iter_cached', None):
@@ -479,7 +478,7 @@ def save_cached(task, broker):
tasks.async_chain(task['chain'], group=group, cached=task['cached'], sync=task['sync'], broker=broker) tasks.async_chain(task['chain'], group=group, cached=task['cached'], sync=task['sync'], broker=broker)
# save the task # save the task
broker.cache.set(task_key, broker.cache.set(task_key,
signing.SignedPackage.dumps(task), SignedPackage.dumps(task),
timeout) timeout)
except Exception as e: except Exception as e:
logger.error(e) logger.error(e)
django_q/status.py
+6 -6
View File
@@ -2,7 +2,7 @@ import socket
from django.utils import timezone from django.utils import timezone
from django_q.brokers import get_broker from django_q.brokers import get_broker
from django_q.conf import Conf, logger from django_q.conf import Conf, logger
import signing from django_q.signing import SignedPackage, BadSignature
class Status(object): class Status(object):
@@ -64,7 +64,7 @@ class Stat(Status):
def save(self): def save(self):
try: try:
self.broker.set_stat(self.key, signing.SignedPackage.dumps(self, True), 3) self.broker.set_stat(self.key, SignedPackage.dumps(self, True), 3)
except Exception as e: except Exception as e:
logger.error(e) logger.error(e)
@@ -83,8 +83,8 @@ class Stat(Status):
pack = broker.get_stat(Stat.get_key(cluster_id)) pack = broker.get_stat(Stat.get_key(cluster_id))
if pack: if pack:
try: try:
return signing.SignedPackage.loads(pack) return SignedPackage.loads(pack)
except signing.BadSignature: except BadSignature:
return None return None
return Status(cluster_id) return Status(cluster_id)
@@ -101,8 +101,8 @@ class Stat(Status):
packs = broker.get_stats('{}:*'.format(Conf.Q_STAT)) or [] packs = broker.get_stats('{}:*'.format(Conf.Q_STAT)) or []
for pack in packs: for pack in packs:
try: try:
stats.append(signing.SignedPackage.loads(pack)) stats.append(SignedPackage.loads(pack))
except signing.BadSignature: except BadSignature:
continue continue
return stats return stats
django_q/tasks.py
+49 -45
View File
@@ -1,4 +1,6 @@
"""Provides task functionality.""" """Provides task functionality."""
# Standard
from time import sleep, time
from multiprocessing import Value from multiprocessing import Value
# django # django
@@ -6,9 +8,7 @@ from django.db import IntegrityError
from django.utils import timezone from django.utils import timezone
# local # local
import time from django_q.signing import SignedPackage
import signing
import cluster
from django_q.conf import Conf, logger from django_q.conf import Conf, logger
from django_q.models import Schedule, Task from django_q.models import Schedule, Task
from django_q.humanhash import uuid from django_q.humanhash import uuid
@@ -48,7 +48,7 @@ def async(func, *args, **kwargs):
# signal it # signal it
pre_enqueue.send(sender="django_q", task=task) pre_enqueue.send(sender="django_q", task=task)
# sign it # sign it
pack = signing.SignedPackage.dumps(task) pack = SignedPackage.dumps(task)
if task.get('sync', False): if task.get('sync', False):
return _sync(pack) return _sync(pack)
# push it # push it
@@ -112,14 +112,14 @@ def result(task_id, wait=0, cached=Conf.CACHED):
""" """
if cached: if cached:
return result_cached(task_id, wait) return result_cached(task_id, wait)
start = time.time() start = time()
while True: while True:
r = Task.get_result(task_id) r = Task.get_result(task_id)
if r: if r:
return r return r
if (time.time() - start) * 1000 >= wait >= 0: if (time() - start) * 1000 >= wait >= 0:
break break
time.sleep(0.01) sleep(0.01)
def result_cached(task_id, wait=0, broker=None): def result_cached(task_id, wait=0, broker=None):
@@ -128,14 +128,14 @@ def result_cached(task_id, wait=0, broker=None):
""" """
if not broker: if not broker:
broker = get_broker() broker = get_broker()
start = time.time() start = time()
while True: while True:
r = broker.cache.get('{}:{}'.format(broker.list_key, task_id)) r = broker.cache.get('{}:{}'.format(broker.list_key, task_id))
if r: if r:
return signing.SignedPackage.loads(r)['result'] return SignedPackage.loads(r)['result']
if (time.time() - start) * 1000 >= wait >= 0: if (time() - start) * 1000 >= wait >= 0:
break break
time.sleep(0.01) sleep(0.01)
def result_group(group_id, failures=False, wait=0, count=None, cached=Conf.CACHED): def result_group(group_id, failures=False, wait=0, count=None, cached=Conf.CACHED):
@@ -150,19 +150,19 @@ def result_group(group_id, failures=False, wait=0, count=None, cached=Conf.CACHE
""" """
if cached: if cached:
return result_group_cached(group_id, failures, wait, count) return result_group_cached(group_id, failures, wait, count)
start = time.time() start = time()
if count: if count:
while True: while True:
if count_group(group_id) == count or wait and (time.time() - start) * 1000 >= wait >= 0: if count_group(group_id) == count or wait and (time() - start) * 1000 >= wait >= 0:
break break
time.sleep(0.01) sleep(0.01)
while True: while True:
r = Task.get_result_group(group_id, failures) r = Task.get_result_group(group_id, failures)
if r: if r:
return r return r
if (time.time() - start) * 1000 >= wait >= 0: if (time() - start) * 1000 >= wait >= 0:
break break
time.sleep(0.01) sleep(0.01)
def result_group_cached(group_id, failures=False, wait=0, count=None, broker=None): def result_group_cached(group_id, failures=False, wait=0, count=None, broker=None):
@@ -171,24 +171,24 @@ def result_group_cached(group_id, failures=False, wait=0, count=None, broker=Non
""" """
if not broker: if not broker:
broker = get_broker() broker = get_broker()
start = time.time() start = time()
if count: if count:
while True: while True:
if count_group_cached(group_id) == count or wait and (time.time() - start) * 1000 >= wait > 0: if count_group_cached(group_id) == count or wait and (time() - start) * 1000 >= wait > 0:
break break
time.sleep(0.01) sleep(0.01)
while True: while True:
group_list = broker.cache.get('{}:{}:keys'.format(broker.list_key, group_id)) group_list = broker.cache.get('{}:{}:keys'.format(broker.list_key, group_id))
if group_list: if group_list:
result_list = [] result_list = []
for task_key in group_list: for task_key in group_list:
task = signing.SignedPackage.loads(broker.cache.get(task_key)) task = SignedPackage.loads(broker.cache.get(task_key))
if task['success'] or failures: if task['success'] or failures:
result_list.append(task['result']) result_list.append(task['result'])
return result_list return result_list
if (time.time() - start) * 1000 >= wait >= 0: if (time() - start) * 1000 >= wait >= 0:
break break
time.sleep(0.01) sleep(0.01)
def fetch(task_id, wait=0, cached=Conf.CACHED): def fetch(task_id, wait=0, cached=Conf.CACHED):
@@ -205,14 +205,14 @@ def fetch(task_id, wait=0, cached=Conf.CACHED):
""" """
if cached: if cached:
return fetch_cached(task_id, wait) return fetch_cached(task_id, wait)
start = time.time() start = time()
while True: while True:
t = Task.get_task(task_id) t = Task.get_task(task_id)
if t: if t:
return t return t
if (time.time() - start) * 1000 >= wait >= 0: if (time() - start) * 1000 >= wait >= 0:
break break
time.sleep(0.01) sleep(0.01)
def fetch_cached(task_id, wait=0, broker=None): def fetch_cached(task_id, wait=0, broker=None):
@@ -221,11 +221,11 @@ def fetch_cached(task_id, wait=0, broker=None):
""" """
if not broker: if not broker:
broker = get_broker() broker = get_broker()
start = time.time() start = time()
while True: while True:
r = broker.cache.get('{}:{}'.format(broker.list_key, task_id)) r = broker.cache.get('{}:{}'.format(broker.list_key, task_id))
if r: if r:
task = signing.SignedPackage.loads(r) task = SignedPackage.loads(r)
t = Task(id=task['id'], t = Task(id=task['id'],
name=task['name'], name=task['name'],
func=task['func'], func=task['func'],
@@ -237,9 +237,9 @@ def fetch_cached(task_id, wait=0, broker=None):
result=task['result'], result=task['result'],
success=task['success']) success=task['success'])
return t return t
if (time.time() - start) * 1000 >= wait >= 0: if (time() - start) * 1000 >= wait >= 0:
break break
time.sleep(0.01) sleep(0.01)
def fetch_group(group_id, failures=True, wait=0, count=None, cached=Conf.CACHED): def fetch_group(group_id, failures=True, wait=0, count=None, cached=Conf.CACHED):
@@ -253,19 +253,19 @@ def fetch_group(group_id, failures=True, wait=0, count=None, cached=Conf.CACHED)
""" """
if cached: if cached:
return fetch_group_cached(group_id, failures, wait, count) return fetch_group_cached(group_id, failures, wait, count)
start = time.time() start = time()
if count: if count:
while True: while True:
if count_group(group_id) == count or wait and (time.time() - start) * 1000 >= wait >= 0: if count_group(group_id) == count or wait and (time() - start) * 1000 >= wait >= 0:
break break
time.sleep(0.01) sleep(0.01)
while True: while True:
r = Task.get_task_group(group_id, failures) r = Task.get_task_group(group_id, failures)
if r: if r:
return r return r
if (time.time() - start) * 1000 >= wait >= 0: if (time() - start) * 1000 >= wait >= 0:
break break
time.sleep(0.01) sleep(0.01)
def fetch_group_cached(group_id, failures=True, wait=0, count=None, broker=None): def fetch_group_cached(group_id, failures=True, wait=0, count=None, broker=None):
@@ -274,18 +274,18 @@ def fetch_group_cached(group_id, failures=True, wait=0, count=None, broker=None)
""" """
if not broker: if not broker:
broker = get_broker() broker = get_broker()
start = time.time() start = time()
if count: if count:
while True: while True:
if count_group_cached(group_id) == count or wait and (time.time() - start) * 1000 >= wait >= 0: if count_group_cached(group_id) == count or wait and (time() - start) * 1000 >= wait >= 0:
break break
time.sleep(0.01) sleep(0.01)
while True: while True:
group_list = broker.cache.get('{}:{}:keys'.format(broker.list_key, group_id)) group_list = broker.cache.get('{}:{}:keys'.format(broker.list_key, group_id))
if group_list: if group_list:
task_list = [] task_list = []
for task_key in group_list: for task_key in group_list:
task = signing.SignedPackage.loads(broker.cache.get(task_key)) task = SignedPackage.loads(broker.cache.get(task_key))
if task['success'] or failures: if task['success'] or failures:
t = Task(id=task['id'], t = Task(id=task['id'],
name=task['name'], name=task['name'],
@@ -300,9 +300,9 @@ def fetch_group_cached(group_id, failures=True, wait=0, count=None, broker=None)
success=task['success']) success=task['success'])
task_list.append(t) task_list.append(t)
return task_list return task_list
if (time.time() - start) * 1000 >= wait >= 0: if (time() - start) * 1000 >= wait >= 0:
break break
time.sleep(0.01) sleep(0.01)
def count_group(group_id, failures=False, cached=Conf.CACHED): def count_group(group_id, failures=False, cached=Conf.CACHED):
@@ -332,7 +332,7 @@ def count_group_cached(group_id, failures=False, broker=None):
return len(group_list) return len(group_list)
failure_count = 0 failure_count = 0
for task_key in group_list: for task_key in group_list:
task = signing.SignedPackage.loads(broker.cache.get(task_key)) task = SignedPackage.loads(broker.cache.get(task_key))
if not task['success']: if not task['success']:
failure_count += 1 failure_count += 1
return failure_count return failure_count
@@ -405,7 +405,7 @@ def async_iter(func, args_iter, **kwargs):
options['cached'] = True options['cached'] = True
# save the original arguments # save the original arguments
broker = options['broker'] broker = options['broker']
broker.cache.set('{}:{}:args'.format(broker.list_key, iter_group), signing.SignedPackage.dumps(args_iter)) broker.cache.set('{}:{}:args'.format(broker.list_key, iter_group), SignedPackage.dumps(args_iter))
for args in args_iter: for args in args_iter:
if type(args) is not tuple: if type(args) is not tuple:
args = (args,) args = (args,)
@@ -671,13 +671,17 @@ class Async(object):
def _sync(pack): def _sync(pack):
# Python 2.6 is unable to handle this import on top of the file
# because it creates a circular dependency between tasks and cluster
from django_q.cluster import worker, monitor
"""Simulate a package travelling through the cluster.""" """Simulate a package travelling through the cluster."""
task_queue = Queue() task_queue = Queue()
result_queue = Queue() result_queue = Queue()
task = signing.SignedPackage.loads(pack) task = SignedPackage.loads(pack)
task_queue.put(task) task_queue.put(task)
task_queue.put('STOP') task_queue.put('STOP')
cluster.worker(task_queue, result_queue, Value('f', -1)) worker(task_queue, result_queue, Value('f', -1))
result_queue.put('STOP') result_queue.put('STOP')
cluster.monitor(result_queue) monitor(result_queue)
return task['id'] return task['id']