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

Change imports signing classes explicit

Browse Source
This commit is contained in:
Eagllus
2018-03-13 09:22:09 +01:00
parent ae9392137c
commit 9c39546417
3 changed files with 21 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files
django_q/cluster.py
+6 -6
View File
@@ -21,12 +21,12 @@ from django.utils.translation import ugettext_lazy as _
from django import db from django import db
# Local # Local
import signing
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 +297,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 +456,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 +479,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
+9 -9
View File
@@ -7,7 +7,7 @@ from django.utils import timezone
# local # local
import time import time
import signing from django_q.signing import SignedPackage
import cluster 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
@@ -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
@@ -132,7 +132,7 @@ def result_cached(task_id, wait=0, broker=None):
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.time() - start) * 1000 >= wait >= 0:
break break
time.sleep(0.01) time.sleep(0.01)
@@ -182,7 +182,7 @@ def result_group_cached(group_id, failures=False, wait=0, count=None, broker=Non
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
@@ -225,7 +225,7 @@ def fetch_cached(task_id, wait=0, broker=None):
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'],
@@ -285,7 +285,7 @@ def fetch_group_cached(group_id, failures=True, wait=0, count=None, broker=None)
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'],
@@ -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,)
@@ -674,7 +674,7 @@ def _sync(pack):
"""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)) cluster.worker(task_queue, result_queue, Value('f', -1))