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

Merge pull request #310 from Koed00/python37

Browse Source 
Python 3.7 support and deprecation of Python 2 and older Django versions -
This commit is contained in:
Ilan Steemers
2018-08-14 12:17:52 +02:00
committed by GitHub
parent 2040771e80 239345a5ec
commit 35ba23de89
29 changed files with 254 additions and 326 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.travis.yml
+6 -6
View File
@@ -5,20 +5,20 @@ services:
- mongodb - mongodb
python: python:
- "2.7"
- "3.6" - "3.6"
- "3.7"
env: env:
- DJANGO=2.0 - DJANGO=2.1
- DJANGO=1.11.11 - DJANGO=1.11.11
- DJANGO=1.8.19
matrix: matrix:
exclude: exclude:
- python: "2.7" - python: "3.7"
env: DJANGO=2.0 env: DJANGO=1.11.11
sudo: false sudo: true
dist: xenial
addons: addons:
apt: apt:
README.rst
+7 -7
View File
@@ -26,12 +26,12 @@ Features
Requirements Requirements
~~~~~~~~~~~~ ~~~~~~~~~~~~
- `Django <https://www.djangoproject.com>`__ > = 1.8 - `Django <https://www.djangoproject.com>`__ > = 1.11.11
- `Django-picklefield <https://github.com/gintas/django-picklefield>`__ - `Django-picklefield <https://github.com/gintas/django-picklefield>`__
- `Arrow <https://github.com/crsmithdev/arrow>`__ - `Arrow <https://github.com/crsmithdev/arrow>`__
- `Blessed <https://github.com/jquast/blessed>`__ - `Blessed <https://github.com/jquast/blessed>`__
Tested with: Python 2.7 & 3.6. Django 1.8.19, 1.11.11 and 2.0.x Tested with: Python 3.6. 3.7 Django 1.11.11 and 2.0.x
Brokers Brokers
~~~~~~~ ~~~~~~~
@@ -110,19 +110,19 @@ Check overall statistics with::
Creating Tasks Creating Tasks
~~~~~~~~~~~~~~ ~~~~~~~~~~~~~~
Use `async` from your code to quickly offload tasks: Use `async_task` from your code to quickly offload tasks:
.. code:: python .. code:: python
from django_q.tasks import async, result from django_q.tasks import async_task, result
# create the task # create the task
async('math.copysign', 2, -2) async_task('math.copysign', 2, -2)
# or with a reference # or with a reference
import math.copysign import math.copysign
task_id = async(copysign, 2, -2) task_id = async_task(copysign, 2, -2)
# get the result # get the result
task_result = result(task_id) task_result = result(task_id)
@@ -133,7 +133,7 @@ Use `async` from your code to quickly offload tasks:
# but in most cases you will want to use a hook: # but in most cases you will want to use a hook:
async('math.modf', 2.5, hook='hooks.print_result') async_task('math.modf', 2.5, hook='hooks.print_result')
# hooks.py # hooks.py
def print_result(task): def print_result(task):
django_q/__init__.py
+1 -16
View File
@@ -1,21 +1,6 @@
# import os VERSION = (1, 0, 0)
# import sys
import django
# myPath = os.path.dirname(os.path.abspath(__file__))
# sys.path.insert(0, myPath)
VERSION = (0, 9, 4)
default_app_config = 'django_q.apps.DjangoQConfig' default_app_config = 'django_q.apps.DjangoQConfig'
# root imports will slowly be deprecated.
# please import from the relevant sub modules
if django.VERSION[:2] < (1, 9):
from .tasks import async, schedule, result, result_group, fetch, fetch_group, count_group, delete_group, queue_size
from .models import Task, Schedule, Success, Failure
from .cluster import Cluster
from .status import Stat
from .brokers import get_broker
__all__ = ['conf', 'cluster', 'models', 'tasks'] __all__ = ['conf', 'cluster', 'models', 'tasks']
django_q/admin.py
+10 -11
View File
@@ -2,9 +2,9 @@
from django.contrib import admin from django.contrib import admin
from django.utils.translation import ugettext_lazy as _ from django.utils.translation import ugettext_lazy as _
from django_q.tasks import async
from django_q.models import Success, Failure, Schedule, OrmQ
from django_q.conf import Conf from django_q.conf import Conf
from django_q.models import Success, Failure, Schedule, OrmQ
from django_q.tasks import async_task
class TaskAdmin(admin.ModelAdmin): class TaskAdmin(admin.ModelAdmin):
@@ -19,7 +19,7 @@ class TaskAdmin(admin.ModelAdmin):
'group' 'group'
) )
def has_add_permission(self, request, obj=None): def has_add_permission(self, request):
"""Don't allow adds.""" """Don't allow adds."""
return False return False
@@ -34,14 +34,13 @@ class TaskAdmin(admin.ModelAdmin):
def get_readonly_fields(self, request, obj=None): def get_readonly_fields(self, request, obj=None):
"""Set all fields readonly.""" """Set all fields readonly."""
return list(self.readonly_fields) + \ return list(self.readonly_fields) + [field.name for field in obj._meta.fields]
[field.name for field in obj._meta.fields]
def retry_failed(FailAdmin, request, queryset): def retry_failed(FailAdmin, request, queryset):
"""Submit selected tasks back to the queue.""" """Submit selected tasks back to the queue."""
for task in queryset: for task in queryset:
async(task.func, *task.args or (), hook=task.hook, **task.kwargs or {}) async_task(task.func, *task.args or (), hook=task.hook, **task.kwargs or {})
task.delete() task.delete()
@@ -56,10 +55,10 @@ class FailAdmin(admin.ModelAdmin):
'func', 'func',
'started', 'started',
'stopped', 'stopped',
'result' 'short_result'
) )
def has_add_permission(self, request, obj=None): def has_add_permission(self, request):
"""Don't allow adds.""" """Don't allow adds."""
return False return False
@@ -70,8 +69,7 @@ class FailAdmin(admin.ModelAdmin):
def get_readonly_fields(self, request, obj=None): def get_readonly_fields(self, request, obj=None):
"""Set all fields readonly.""" """Set all fields readonly."""
return list(self.readonly_fields) + \ return list(self.readonly_fields) + [field.name for field in obj._meta.fields]
[field.name for field in obj._meta.fields]
class ScheduleAdmin(admin.ModelAdmin): class ScheduleAdmin(admin.ModelAdmin):
@@ -113,10 +111,11 @@ class QueueAdmin(admin.ModelAdmin):
def get_queryset(self, request): def get_queryset(self, request):
return super(QueueAdmin, self).get_queryset(request).using(Conf.ORM) return super(QueueAdmin, self).get_queryset(request).using(Conf.ORM)
def has_add_permission(self, request, obj=None): def has_add_permission(self, request):
"""Don't allow adds.""" """Don't allow adds."""
return False return False
admin.site.register(Schedule, ScheduleAdmin) admin.site.register(Schedule, ScheduleAdmin)
admin.site.register(Success, TaskAdmin) admin.site.register(Success, TaskAdmin)
admin.site.register(Failure, FailAdmin) admin.site.register(Failure, FailAdmin)
django_q/cluster.py
+18 -23
View File
@@ -4,32 +4,31 @@ from __future__ import division
from __future__ import print_function from __future__ import print_function
from __future__ import unicode_literals from __future__ import unicode_literals
from time import sleep
# external
import arrow
import ast
# Standard # Standard
import importlib import importlib
import signal import signal
import socket import socket
import ast import traceback
from time import sleep
from multiprocessing import Event, Process, Value, current_process
# external
import arrow
# Django # Django
from django import db
from django.utils import timezone from django.utils import timezone
from django.utils.translation import ugettext_lazy as _ from django.utils.translation import ugettext_lazy as _
from django import db from multiprocessing import Event, Process, Value, current_process
# Local # Local
from django_q import tasks from django_q import tasks
from django_q.compat import range from django_q.brokers import get_broker
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
from django_q.models import Task, Success, Schedule from django_q.models import Task, Success, Schedule
from django_q.queues import Queue
from django_q.signals import pre_execute
from django_q.signing import SignedPackage, BadSignature 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.signals import pre_execute
from django_q.queues import Queue
class Cluster(object): class Cluster(object):
@@ -287,7 +286,7 @@ def pusher(task_queue, event, broker=None):
try: try:
task_set = broker.dequeue() task_set = broker.dequeue()
except Exception as e: except Exception as e:
logger.error(e) logger.error(e, traceback.format_exc())
# broker probably crashed. Let the sentinel handle it. # broker probably crashed. Let the sentinel handle it.
sleep(10) sleep(10)
break break
@@ -298,7 +297,7 @@ def pusher(task_queue, event, broker=None):
try: try:
task = SignedPackage.loads(task[1]) task = SignedPackage.loads(task[1])
except (TypeError, BadSignature) as e: except (TypeError, BadSignature) as e:
logger.error(e) logger.error(e, traceback.format_exc())
broker.fail(ack_id) broker.fail(ack_id)
continue continue
task['ack_id'] = ack_id task['ack_id'] = ack_id
@@ -366,8 +365,6 @@ def worker(task_queue, result_queue, timer, timeout=Conf.TIMEOUT):
result = (e, False) result = (e, False)
if error_reporter: if error_reporter:
error_reporter.report() error_reporter.report()
if rollbar:
rollbar.report_exc_info()
# We're still going # We're still going
if not result: if not result:
db.close_old_connections() db.close_old_connections()
@@ -380,11 +377,9 @@ def worker(task_queue, result_queue, timer, timeout=Conf.TIMEOUT):
res = f(*task['args'], **task['kwargs']) res = f(*task['args'], **task['kwargs'])
result = (res, True) result = (res, True)
except Exception as e: except Exception as e:
result = ('{}'.format(e), False) result = ('{} : {}'.format(e, traceback.format_exc()), False)
if error_reporter: if error_reporter:
error_reporter.report() error_reporter.report()
if rollbar:
rollbar.report_exc_info()
# Process result # Process result
task['result'] = result[0] task['result'] = result[0]
task['success'] = result[1] task['success'] = result[1]
@@ -405,7 +400,7 @@ def save_task(task, broker):
# SAVE LIMIT < 0 : Don't save success # 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 return
# async next in a chain # enqueues next in a chain
if task.get('chain', None): if task.get('chain', None):
tasks.async_chain(task['chain'], group=task['group'], cached=task['cached'], sync=task['sync'], broker=broker) 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 # SAVE LIMIT > 0: Prune database, SAVE_LIMIT 0: No pruning
@@ -473,7 +468,7 @@ def save_cached(task, broker):
# save the group list # save the group list
group_list.append(task_key) group_list.append(task_key)
broker.cache.set(group_key, group_list, timeout) broker.cache.set(group_key, group_list, timeout)
# async next in a chain # async_task next in a chain
if task.get('chain', None): if task.get('chain', None):
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
@@ -536,7 +531,7 @@ def scheduler(broker=None):
q_options['broker'] = broker q_options['broker'] = broker
q_options['group'] = q_options.get('group', s.name or s.id) q_options['group'] = q_options.get('group', s.name or s.id)
kwargs['q_options'] = q_options kwargs['q_options'] = q_options
s.task = tasks.async(s.func, *args, **kwargs) s.task = tasks.async_task(s.func, *args, **kwargs)
# log it # log it
if not s.task: if not s.task:
logger.error( logger.error(
django_q/compat.py
-14
View File
@@ -1,14 +0,0 @@
from __future__ import absolute_import
"""
Compatibility layer.
Intentionally replaces use of python-future
"""
# https://github.com/Koed00/django-q/issues/4
try:
range = xrange
except NameError:
range = range
django_q/conf.py
-16
View File
@@ -150,9 +150,6 @@ class Conf(object):
# The redis stats key # The redis stats key
Q_STAT = 'django_q:{}:cluster'.format(PREFIX) Q_STAT = 'django_q:{}:cluster'.format(PREFIX)
# Optional rollbar key
ROLLBAR = conf.get('rollbar', {})
# Optional error reporting setup # Optional error reporting setup
ERROR_REPORTER = conf.get('error_reporter', {}) ERROR_REPORTER = conf.get('error_reporter', {})
@@ -190,19 +187,6 @@ if not logger.handlers:
logger.addHandler(handler) logger.addHandler(handler)
# rollbar
if Conf.ROLLBAR:
rollbar_conf = deepcopy(Conf.ROLLBAR)
try:
import rollbar
rollbar.init(rollbar_conf.pop('access_token'), environment=rollbar_conf.pop('environment'), **rollbar_conf)
except ImportError:
rollbar = None
else:
rollbar = None
# Error Reporting Interface # Error Reporting Interface
class ErrorReporter(object): class ErrorReporter(object):
django_q/models.py
+6 -3
View File
@@ -1,8 +1,7 @@
from django import get_version from django import get_version
try: from django.template.defaultfilters import truncatechars
from django.urls import reverse from django.urls import reverse
except ImportError: # Django < 1.10
from django.core.urlresolvers import reverse
from django.utils.html import format_html from django.utils.html import format_html
from django.utils.translation import ugettext_lazy as _ from django.utils.translation import ugettext_lazy as _
from django.db import models from django.db import models
@@ -82,6 +81,10 @@ class Task(models.Model):
def time_taken(self): def time_taken(self):
return (self.stopped - self.started).total_seconds() return (self.stopped - self.started).total_seconds()
@property
def short_result(self):
return truncatechars(self.result, 100)
def __unicode__(self): def __unicode__(self):
return u'{}'.format(self.name or self.id) return u'{}'.format(self.name or self.id)
django_q/tasks.py
+23 -21
View File
@@ -1,26 +1,28 @@
"""Provides task functionality.""" """Provides task functionality."""
# Standard # Standard
from time import sleep, time from time import sleep, time
from multiprocessing import Value
# django # django
from django.db import IntegrityError from django.db import IntegrityError
from django.utils import timezone from django.utils import timezone
from multiprocessing import Value
# local
from django_q.signing import SignedPackage
from django_q.conf import Conf, logger
from django_q.models import Schedule, Task
from django_q.humanhash import uuid
from django_q.brokers import get_broker from django_q.brokers import get_broker
from django_q.signals import pre_enqueue # local
from django_q.cluster import worker, monitor
from django_q.conf import Conf, logger
from django_q.humanhash import uuid
from django_q.models import Schedule, Task
from django_q.queues import Queue from django_q.queues import Queue
from django_q.signals import pre_enqueue
from django_q.signing import SignedPackage
def async(func, *args, **kwargs): def async_task(func, *args, **kwargs):
"""Queue a task for the cluster.""" """Queue a task for the cluster."""
keywords = kwargs.copy() keywords = kwargs.copy()
opt_keys = ('hook', 'group', 'save', 'sync', 'cached', 'ack_failure', 'iter_count', 'iter_cached', 'chain', 'broker') opt_keys = (
'hook', 'group', 'save', 'sync', 'cached', 'ack_failure', 'iter_count', 'iter_cached', 'chain', 'broker')
q_options = keywords.pop('q_options', {}) q_options = keywords.pop('q_options', {})
# get an id # get an id
tag = uuid() tag = uuid()
@@ -392,7 +394,7 @@ def queue_size(broker=None):
def async_iter(func, args_iter, **kwargs): def async_iter(func, args_iter, **kwargs):
""" """
async a function with iterable arguments enqueues a function with iterable arguments
""" """
iter_count = len(args_iter) iter_count = len(args_iter)
iter_group = uuid()[1] iter_group = uuid()[1]
@@ -409,15 +411,15 @@ def async_iter(func, args_iter, **kwargs):
broker = options['broker'] broker = options['broker']
broker.cache.set('{}:{}:args'.format(broker.list_key, iter_group), 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 not isinstance(args, tuple):
args = (args,) args = (args,)
async(func, *args, **options) async_task(func, *args, **options)
return iter_group return iter_group
def async_chain(chain, group=None, cached=Conf.CACHED, sync=Conf.SYNC, broker=None): def async_chain(chain, group=None, cached=Conf.CACHED, sync=Conf.SYNC, broker=None):
""" """
async a chain of tasks enqueues a chain of tasks
the chain must be in the format [(func,(args),{kwargs}),(func,(args),{kwargs})] the chain must be in the format [(func,(args),{kwargs}),(func,(args),{kwargs})]
""" """
if not group: if not group:
@@ -436,7 +438,7 @@ def async_chain(chain, group=None, cached=Conf.CACHED, sync=Conf.SYNC, broker=No
kwargs['cached'] = cached kwargs['cached'] = cached
kwargs['sync'] = sync kwargs['sync'] = sync
kwargs['broker'] = broker or get_broker() kwargs['broker'] = broker or get_broker()
async(task[0], *args, **kwargs) async_task(task[0], *args, **kwargs)
return group return group
@@ -518,7 +520,7 @@ class Chain(object):
def append(self, func, *args, **kwargs): def append(self, func, *args, **kwargs):
""" """
add a task to the chain add a task to the chain
takes the same parameters as async() takes the same parameters as async_task()
""" """
self.chain.append((func, args, kwargs)) self.chain.append((func, args, kwargs))
# remove existing results # remove existing results
@@ -573,7 +575,7 @@ class Chain(object):
return len(self.chain) return len(self.chain)
class Async(object): class AsyncTask(object):
""" """
an async task an async task
""" """
@@ -647,7 +649,7 @@ class Async(object):
return self.kwargs.get(key, default) return self.kwargs.get(key, default)
def run(self): def run(self):
self.id = async(self.func, *self.args, **self.kwargs) self.id = async_task(self.func, *self.args, **self.kwargs)
self.started = True self.started = True
return self.id return self.id
@@ -673,10 +675,6 @@ 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()
@@ -686,4 +684,8 @@ def _sync(pack):
worker(task_queue, result_queue, Value('f', -1)) worker(task_queue, result_queue, Value('f', -1))
result_queue.put('STOP') result_queue.put('STOP')
monitor(result_queue) monitor(result_queue)
task_queue.close()
task_queue.join_thread()
result_queue.close()
result_queue.join_thread()
return task['id'] return task['id']
django_q/tests/test_admin.py
-3
View File
@@ -1,7 +1,4 @@
try:
from django.urls import reverse from django.urls import reverse
except ImportError: # Django < 1.10
from django.core.urlresolvers import reverse
from django.utils import timezone from django.utils import timezone
import pytest import pytest
django_q/tests/test_brokers.py
+6 -7
View File
@@ -5,7 +5,6 @@ import pytest
import redis import redis
from django_q.brokers import get_broker, Broker from django_q.brokers import get_broker, Broker
from django_q.compat import range
from django_q.conf import Conf from django_q.conf import Conf
from django_q.humanhash import uuid from django_q.humanhash import uuid
@@ -63,7 +62,7 @@ def test_disque(monkeypatch):
assert broker.info() is not None assert broker.info() is not None
# clear before we start # clear before we start
broker.delete_queue() broker.delete_queue()
# enqueue # async_task
broker.enqueue('test') broker.enqueue('test')
assert broker.queue_size() == 1 assert broker.queue_size() == 1
# dequeue # dequeue
@@ -127,7 +126,7 @@ def test_ironmq(monkeypatch):
# clear before we start # clear before we start
broker.purge_queue() broker.purge_queue()
assert broker.queue_size() == 0 assert broker.queue_size() == 0
# enqueue # async_task
broker.enqueue('test') broker.enqueue('test')
# dequeue # dequeue
task = broker.dequeue()[0] task = broker.dequeue()[0]
@@ -136,7 +135,7 @@ def test_ironmq(monkeypatch):
assert broker.dequeue() is None assert broker.dequeue() is None
# Retry test # Retry test
# monkeypatch.setattr(Conf, 'RETRY', 1) # monkeypatch.setattr(Conf, 'RETRY', 1)
# broker.enqueue('test') # broker.async_task('test')
# assert broker.dequeue() is not None # assert broker.dequeue() is not None
# sleep(3) # sleep(3)
# assert broker.dequeue() is not None # assert broker.dequeue() is not None
@@ -180,7 +179,7 @@ def canceled_sqs(monkeypatch):
assert broker.ping() is True assert broker.ping() is True
assert broker.info() is not None assert broker.info() is not None
assert broker.queue_size() == 0 assert broker.queue_size() == 0
# enqueue # async_task
broker.enqueue('test') broker.enqueue('test')
# dequeue # dequeue
task = broker.dequeue()[0] task = broker.dequeue()[0]
@@ -240,7 +239,7 @@ def test_orm(monkeypatch):
assert broker.info() is not None assert broker.info() is not None
# clear before we start # clear before we start
broker.delete_queue() broker.delete_queue()
# enqueue # async_task
broker.enqueue('test') broker.enqueue('test')
assert broker.queue_size() == 1 assert broker.queue_size() == 1
# dequeue # dequeue
@@ -297,7 +296,7 @@ def test_mongo(monkeypatch):
assert broker.info() is not None assert broker.info() is not None
# clear before we start # clear before we start
broker.delete_queue() broker.delete_queue()
# enqueue # async_task
broker.enqueue('test') broker.enqueue('test')
assert broker.queue_size() == 1 assert broker.queue_size() == 1
# dequeue # dequeue
django_q/tests/test_cached.py
+14 -15
View File
@@ -3,10 +3,9 @@ from multiprocessing import Event, Value
import pytest import pytest
from django_q.cluster import pusher, worker, monitor from django_q.cluster import pusher, worker, monitor
from django_q.compat import range
from django_q.conf import Conf from django_q.conf import Conf
from django_q.tasks import async, result, fetch, count_group, result_group, fetch_group, delete_group, delete_cached, \ from django_q.tasks import async_task, result, fetch, count_group, result_group, fetch_group, delete_group, delete_cached, \
async_iter, Chain, async_chain, Iter, Async async_iter, Chain, async_chain, Iter, AsyncTask
from django_q.brokers import get_broker from django_q.brokers import get_broker
from django_q.queues import Queue from django_q.queues import Queue
@@ -23,13 +22,13 @@ def test_cached(broker):
broker.cache.clear() broker.cache.clear()
group = 'cache_test' group = 'cache_test'
# queue the tests # queue the tests
task_id = async('math.copysign', 1, -1, cached=True, broker=broker) task_id = async_task('math.copysign', 1, -1, cached=True, broker=broker)
async('math.copysign', 1, -1, cached=True, broker=broker, group=group) async_task('math.copysign', 1, -1, cached=True, broker=broker, group=group)
async('math.copysign', 1, -1, cached=True, broker=broker, group=group) async_task('math.copysign', 1, -1, cached=True, broker=broker, group=group)
async('math.copysign', 1, -1, cached=True, broker=broker, group=group) async_task('math.copysign', 1, -1, cached=True, broker=broker, group=group)
async('math.copysign', 1, -1, cached=True, broker=broker, group=group) async_task('math.copysign', 1, -1, cached=True, broker=broker, group=group)
async('math.copysign', 1, -1, cached=True, broker=broker, group=group) async_task('math.copysign', 1, -1, cached=True, broker=broker, group=group)
async('math.popysign', 1, -1, cached=True, broker=broker, group=group) async_task('math.popysign', 1, -1, cached=True, broker=broker, group=group)
iter_id = async_iter('math.floor', [i for i in range(10)], cached=True) iter_id = async_iter('math.floor', [i for i in range(10)], cached=True)
# test wait on cache # test wait on cache
# test wait timeout # test wait timeout
@@ -145,10 +144,10 @@ def test_chain(broker):
@pytest.mark.django_db @pytest.mark.django_db
def test_async_class(broker, monkeypatch): def test_asynctask_class(broker, monkeypatch):
broker.purge_queue() broker.purge_queue()
broker.cache.clear() broker.cache.clear()
a = Async('math.copysign') a = AsyncTask('math.copysign')
assert a.func == 'math.copysign' assert a.func == 'math.copysign'
a.args = (1, -1) a.args = (1, -1)
assert a.started is False assert a.started is False
@@ -162,11 +161,11 @@ def test_async_class(broker, monkeypatch):
assert a.result() == -1 assert a.result() == -1
assert a.fetch().result == -1 assert a.fetch().result == -1
# again with kwargs # again with kwargs
a = Async('math.copysign', 1, -1, cached=True, sync=True, broker=broker) a = AsyncTask('math.copysign', 1, -1, cached=True, sync=True, broker=broker)
a.run() a.run()
assert a.result() == -1 assert a.result() == -1
# with q_options # with q_options
a = Async('math.copysign', 1, -1, q_options={'cached': True, 'sync': False, 'broker': broker}) a = AsyncTask('math.copysign', 1, -1, q_options={'cached': True, 'sync': False, 'broker': broker})
assert a.sync is False assert a.sync is False
a.sync = True a.sync = True
assert a.kwargs['q_options']['sync'] is True assert a.kwargs['q_options']['sync'] is True
@@ -185,6 +184,6 @@ def test_async_class(broker, monkeypatch):
# global overrides # global overrides
monkeypatch.setattr(Conf, 'SYNC', True) monkeypatch.setattr(Conf, 'SYNC', True)
monkeypatch.setattr(Conf, 'CACHED', True) monkeypatch.setattr(Conf, 'CACHED', True)
a = Async('math.floor', 1.5) a = AsyncTask('math.floor', 1.5)
a.run() a.run()
assert a.result() == 1 assert a.result() == 1
django_q/tests/test_cluster.py
+22 -23
View File
@@ -11,9 +11,8 @@ myPath = os.path.dirname(os.path.abspath(__file__))
sys.path.insert(0, myPath + '/../') sys.path.insert(0, myPath + '/../')
from django_q.cluster import Cluster, Sentinel, pusher, worker, monitor, save_task from django_q.cluster import Cluster, Sentinel, pusher, worker, monitor, save_task
from django_q.compat import range
from django_q.humanhash import DEFAULT_WORDLIST, uuid from django_q.humanhash import DEFAULT_WORDLIST, uuid
from django_q.tasks import fetch, fetch_group, async, result, result_group, count_group, delete_group, queue_size from django_q.tasks import fetch, fetch_group, async_task, result, result_group, count_group, delete_group, queue_size
from django_q.models import Task, Success from django_q.models import Task, Success
from django_q.conf import Conf from django_q.conf import Conf
from django_q.status import Stat from django_q.status import Stat
@@ -42,7 +41,7 @@ def test_redis_connection(broker):
@pytest.mark.django_db @pytest.mark.django_db
def test_sync(broker): def test_sync(broker):
task = async('django_q.tests.tasks.count_letters', DEFAULT_WORDLIST, broker=broker, sync=True) task = async_task('django_q.tests.tasks.count_letters', DEFAULT_WORDLIST, broker=broker, sync=True)
assert result(task) == 1506 assert result(task) == 1506
@@ -82,7 +81,7 @@ def test_sentinel():
def test_cluster(broker): def test_cluster(broker):
broker.list_key = 'cluster_test:q' broker.list_key = 'cluster_test:q'
broker.delete_queue() broker.delete_queue()
task = async('django_q.tests.tasks.count_letters', DEFAULT_WORDLIST, broker=broker) task = async_task('django_q.tests.tasks.count_letters', DEFAULT_WORDLIST, broker=broker)
assert broker.queue_size() == 1 assert broker.queue_size() == 1
task_queue = Queue() task_queue = Queue()
assert task_queue.qsize() == 0 assert task_queue.qsize() == 0
@@ -109,31 +108,31 @@ def test_cluster(broker):
@pytest.mark.django_db @pytest.mark.django_db
def test_async(broker, admin_user): def test_enqueue(broker, admin_user):
broker.list_key = 'cluster_test:q' broker.list_key = 'cluster_test:q'
broker.delete_queue() broker.delete_queue()
a = async('django_q.tests.tasks.count_letters', DEFAULT_WORDLIST, hook='django_q.tests.test_cluster.assert_result', a = async_task('django_q.tests.tasks.count_letters', DEFAULT_WORDLIST, hook='django_q.tests.test_cluster.assert_result',
broker=broker) broker=broker)
b = async('django_q.tests.tasks.count_letters2', WordClass(), hook='django_q.tests.test_cluster.assert_result', b = async_task('django_q.tests.tasks.count_letters2', WordClass(), hook='django_q.tests.test_cluster.assert_result',
broker=broker) broker=broker)
# unknown argument # unknown argument
c = async('django_q.tests.tasks.count_letters', DEFAULT_WORDLIST, 'oneargumentoomany', c = async_task('django_q.tests.tasks.count_letters', DEFAULT_WORDLIST, 'oneargumentoomany',
hook='django_q.tests.test_cluster.assert_bad_result', broker=broker) hook='django_q.tests.test_cluster.assert_bad_result', broker=broker)
# unknown function # unknown function
d = async('django_q.tests.tasks.does_not_exist', WordClass(), hook='django_q.tests.test_cluster.assert_bad_result', d = async_task('django_q.tests.tasks.does_not_exist', WordClass(), hook='django_q.tests.test_cluster.assert_bad_result',
broker=broker) broker=broker)
# function without result # function without result
e = async('django_q.tests.tasks.countdown', 100000, broker=broker) e = async_task('django_q.tests.tasks.countdown', 100000, broker=broker)
# function as instance # function as instance
f = async(multiply, 753, 2, hook=assert_result, broker=broker) f = async_task(multiply, 753, 2, hook=assert_result, broker=broker)
# model as argument # model as argument
g = async('django_q.tests.tasks.get_task_name', Task(name='John'), broker=broker) g = async_task('django_q.tests.tasks.get_task_name', Task(name='John'), broker=broker)
# args,kwargs, group and broken hook # args,kwargs, group and broken hook
h = async('django_q.tests.tasks.word_multiply', 2, word='django', hook='fail.me', broker=broker) h = async_task('django_q.tests.tasks.word_multiply', 2, word='django', hook='fail.me', broker=broker)
# args unpickle test # args unpickle test
j = async('django_q.tests.tasks.get_user_id', admin_user, broker=broker, group='test_j') j = async_task('django_q.tests.tasks.get_user_id', admin_user, broker=broker, group='test_j')
# q_options and save opt_out test # q_options and save opt_out test
k = async('django_q.tests.tasks.get_user_id', admin_user, k = async_task('django_q.tests.tasks.get_user_id', admin_user,
q_options={'broker': broker, 'group': 'test_k', 'save': False, 'timeout': 90}) q_options={'broker': broker, 'group': 'test_k', 'save': False, 'timeout': 90})
# check if everything has a task id # check if everything has a task id
assert isinstance(a, str) assert isinstance(a, str)
@@ -249,7 +248,7 @@ def test_timeout(broker):
# set up the Sentinel # set up the Sentinel
broker.list_key = 'timeout_test:q' broker.list_key = 'timeout_test:q'
broker.purge_queue() broker.purge_queue()
async('django_q.tests.tasks.count_forever', broker=broker) async_task('django_q.tests.tasks.count_forever', broker=broker)
start_event = Event() start_event = Event()
stop_event = Event() stop_event = Event()
# Set a timer to stop the Sentinel # Set a timer to stop the Sentinel
@@ -265,7 +264,7 @@ def test_timeout(broker):
def test_timeout_override(broker): def test_timeout_override(broker):
# set up the Sentinel # set up the Sentinel
broker.list_key = 'timeout_override_test:q' broker.list_key = 'timeout_override_test:q'
async('django_q.tests.tasks.count_forever', broker=broker, timeout=1) async_task('django_q.tests.tasks.count_forever', broker=broker, timeout=1)
start_event = Event() start_event = Event()
stop_event = Event() stop_event = Event()
# Set a timer to stop the Sentinel # Set a timer to stop the Sentinel
@@ -281,9 +280,9 @@ def test_timeout_override(broker):
def test_recycle(broker, monkeypatch): def test_recycle(broker, monkeypatch):
# set up the Sentinel # set up the Sentinel
broker.list_key = 'test_recycle_test:q' broker.list_key = 'test_recycle_test:q'
async('django_q.tests.tasks.multiply', 2, 2, broker=broker) async_task('django_q.tests.tasks.multiply', 2, 2, broker=broker)
async('django_q.tests.tasks.multiply', 2, 2, broker=broker) async_task('django_q.tests.tasks.multiply', 2, 2, broker=broker)
async('django_q.tests.tasks.multiply', 2, 2, broker=broker) async_task('django_q.tests.tasks.multiply', 2, 2, broker=broker)
start_event = Event() start_event = Event()
stop_event = Event() stop_event = Event()
# override settings # override settings
@@ -295,8 +294,8 @@ def test_recycle(broker, monkeypatch):
assert start_event.is_set() assert start_event.is_set()
assert s.status() == Conf.STOPPED assert s.status() == Conf.STOPPED
assert s.reincarnations == 1 assert s.reincarnations == 1
async('django_q.tests.tasks.multiply', 2, 2, broker=broker) async_task('django_q.tests.tasks.multiply', 2, 2, broker=broker)
async('django_q.tests.tasks.multiply', 2, 2, broker=broker) async_task('django_q.tests.tasks.multiply', 2, 2, broker=broker)
task_queue = Queue() task_queue = Queue()
result_queue = Queue() result_queue = Queue()
# push two tasks # push two tasks
@@ -318,7 +317,7 @@ def test_recycle(broker, monkeypatch):
@pytest.mark.django_db @pytest.mark.django_db
def test_bad_secret(broker, monkeypatch): def test_bad_secret(broker, monkeypatch):
broker.list_key = 'test_bad_secret:q' broker.list_key = 'test_bad_secret:q'
async('math.copysign', 1, -1, broker=broker) async_task('math.copysign', 1, -1, broker=broker)
stop_event = Event() stop_event = Event()
stop_event.set() stop_event.set()
start_event = Event() start_event = Event()
django_q/tests/test_monitor.py
+2 -3
View File
@@ -1,9 +1,8 @@
import pytest import pytest
from django_q.tasks import async from django_q.tasks import async_task
from django_q.brokers import get_broker from django_q.brokers import get_broker
from django_q.cluster import Cluster from django_q.cluster import Cluster
from django_q.compat import range
from django_q.monitor import monitor, info from django_q.monitor import monitor, info
from django_q.status import Stat from django_q.status import Stat
from django_q.conf import Conf from django_q.conf import Conf
@@ -46,4 +45,4 @@ def test_info():
def do_sync(): def do_sync():
async('django_q.tests.tasks.countdown', 1, sync=True, save=True) async_task('django_q.tests.tasks.countdown', 1, sync=True, save=True)
docs/brokers.rst
+1 -1
View File
@@ -140,7 +140,7 @@ You can override this class if you want to contribute and support your own broke
.. py:class:: Broker .. py:class:: Broker
.. py:method:: enqueue(task) .. py:method:: async_task(task)
Sends a task package to the broker queue and returns a tracking id if available. Sends a task package to the broker queue and returns a tracking id if available.
docs/chain.rst
+2 -2
View File
@@ -6,7 +6,7 @@ Sometimes you want to run tasks sequentially. For that you can use the :func:`as
.. code-block:: python .. code-block:: python
# Async a chain of tasks # async a chain of tasks
from django_q.tasks import async_chain, result_group from django_q.tasks import async_chain, result_group
# the chain must be in the format # the chain must be in the format
@@ -63,7 +63,7 @@ Reference
.. py:method:: append(func, *args, **kwargs) .. py:method:: append(func, *args, **kwargs)
Append a task to the chain. Takes the same arguments as :func:`async` Append a task to the chain. Takes the same arguments as :func:`async_task`
:return: the current number of tasks in the chain :return: the current number of tasks in the chain
:rtype: int :rtype: int
docs/conf.py
+2 -2
View File
@@ -71,9 +71,9 @@ author = 'Ilan Steemers'
# built documents. # built documents.
# #
# The short X.Y version. # The short X.Y version.
version = '0.9' version = '1.0'
# The full version, including alpha/beta/rc tags. # The full version, including alpha/beta/rc tags.
release = '0.9.4' release = '1.0.0'
# The language for content autogenerated by Sphinx. Refer to documentation # The language for content autogenerated by Sphinx. Refer to documentation
# for a list of supported languages. # for a list of supported languages.
docs/configure.rst
+2 -21
View File
@@ -64,7 +64,7 @@ Set this to something that makes sense for your project. Can be overridden for i
ack_failures ack_failures
~~~~~~~~~~~~ ~~~~~~~~~~~~
When set to ``True``, also acknowledge unsuccessful tasks. This causes failed tasks to be considered as successful deliveries, thereby removing them from the task queue. Can also be set per-task by passing the ``ack_failure`` option to :func:`async`. Defaults to ``False``. When set to ``True``, also acknowledge unsuccessful tasks. This causes failed tasks to be considered as successful deliveries, thereby removing them from the task queue. Can also be set per-task by passing the ``ack_failure`` option to :func:`async_task`. Defaults to ``False``.
.. _retry: .. _retry:
@@ -101,7 +101,7 @@ Guard loop sleep in seconds, must be greater than 0 and less than 60.
sync sync
~~~~ ~~~~
When set to ``True`` this configuration option forces all :func:`async` calls to be run with ``sync=True``. When set to ``True`` this configuration option forces all :func:`async_task` calls to be run with ``sync=True``.
Effectively making everything synchronous. Useful for testing. Defaults to ``False``. Effectively making everything synchronous. Useful for testing. Defaults to ``False``.
.. _queue_limit: .. _queue_limit:
@@ -380,25 +380,6 @@ To enable installed error reporters, you must provide the configuration settings
For more information on error reporters and developing error reporting plugins for Django Q, see :doc:`errors<errors>`. For more information on error reporters and developing error reporting plugins for Django Q, see :doc:`errors<errors>`.
rollbar
~~~~~~~
You can redirect worker exceptions directly to your `Rollbar <https://rollbar.com/>`__ dashboard by installing the python notifier with ``pip install rollbar`` and adding this configuration dictionary to your config::
# rollbar config
Q_CLUSTER = {
'rollbar': {
'access_token': '32we33a92a5224jiww8982',
'environment': 'Django-Q'
}
}
Please check the Pyrollbar `configuration reference <https://github.com/rollbar/pyrollbar#configuration-reference>`__ for more options.
Note that you will need a `Rollbar <https://rollbar.com/>`__ account and access token to use this feature.
.. note::
The ``rollbar`` setting is included for backwards compatibility, for those who utilized rollbar configuration before the ``error_reporter`` interface was introduced. Note that Rollbar support can be configured either via the ``rollbar`` setting, or via the ``django-q-rollbar`` package and enabled via the ``error_reporter`` setting above.
cpu_affinity cpu_affinity
~~~~~~~~~~~~ ~~~~~~~~~~~~
docs/examples.rst
+26 -26
View File
@@ -12,14 +12,14 @@ Sending an email can take a while so why not queue it:
# Welcome mail with follow up example # Welcome mail with follow up example
from datetime import timedelta from datetime import timedelta
from django.utils import timezone from django.utils import timezone
from django_q.tasks import async, schedule from django_q.tasks import async_task, schedule
from django_q.models import Schedule from django_q.models import Schedule
def welcome_mail(user): def welcome_mail(user):
msg = 'Welcome to our website' msg = 'Welcome to our website'
# send this message right away # send this message right away
async('django.core.mail.send_mail', async_task('django.core.mail.send_mail',
'Welcome', 'Welcome',
msg, msg,
'from@example.com', 'from@example.com',
@@ -51,7 +51,7 @@ A good place to use async tasks are Django's model signals. You don't want to de
from django.contrib.auth.models import User from django.contrib.auth.models import User
from django.db.models.signals import pre_save from django.db.models.signals import pre_save
from django.dispatch import receiver from django.dispatch import receiver
from django_q.tasks import async from django_q.tasks import async_task
# set up the pre_save signal for our user # set up the pre_save signal for our user
@receiver(pre_save, sender=User) @receiver(pre_save, sender=User)
@@ -64,7 +64,7 @@ A good place to use async tasks are Django's model signals. You don't want to de
# has his email changed? # has his email changed?
if not user.email == instance.email: if not user.email == instance.email:
# tell everyone # tell everyone
async('tasks.inform_everyone', instance) async_task('tasks.inform_everyone', instance)
The task will send a message to everyone else informing them that the users email address has changed. Note that this adds almost no overhead to the save action: The task will send a message to everyone else informing them that the users email address has changed. Note that this adds almost no overhead to the save action:
@@ -87,7 +87,7 @@ The task will send a message to everyone else informing them that the users emai
for u in User.objects.exclude(pk=user.pk): for u in User.objects.exclude(pk=user.pk):
msg = 'Dear {}, {} has a new email address: {}' msg = 'Dear {}, {} has a new email address: {}'
msg = msg.format(u.username, user.username, user.email) msg = msg.format(u.username, user.username, user.email)
async('django.core.mail.send_mail', async_task('django.core.mail.send_mail',
'New email', msg, 'from@example.com', [u.email]) 'New email', msg, 'from@example.com', [u.email])
@@ -104,19 +104,19 @@ In this example the user requests a report and we let the cluster do the generat
.. code-block:: python .. code-block:: python
# Report generation with hook example # Report generation with hook example
from django_q.tasks import async from django_q.tasks import async_task
# views.py # views.py
# user requests a report. # user requests a report.
def create_report(request): def create_report(request):
async('tasks.create_html_report', async_task('tasks.create_html_report',
request.user, request.user,
hook='tasks.email_report') hook='tasks.email_report')
.. code-block:: python .. code-block:: python
# tasks.py # tasks.py
from django_q.tasks import async from django_q.tasks import async_task
# report generator # report generator
def create_html_report(user): def create_html_report(user):
@@ -127,14 +127,14 @@ In this example the user requests a report and we let the cluster do the generat
def email_report(task): def email_report(task):
if task.success: if task.success:
# Email the report # Email the report
async('django.core.mail.send_mail', async_task('django.core.mail.send_mail',
'The report you requested', 'The report you requested',
task.result, task.result,
'from@example.com', 'from@example.com',
task.args[0].email) task.args[0].email)
else: else:
# Tell the admins something went wrong # Tell the admins something went wrong
async('django.core.mail.mail_admins', async_task('django.core.mail.mail_admins',
'Report generation failed', 'Report generation failed',
task.result) task.result)
@@ -152,12 +152,12 @@ here's an example of how you can have Django Q take care of your indexes in real
from .models import Document from .models import Document
from django.db.models.signals import post_save from django.db.models.signals import post_save
from django.dispatch import receiver from django.dispatch import receiver
from django_q.tasks import async from django_q.tasks import async_task
# hook up the post save handler # hook up the post save handler
@receiver(post_save, sender=Document) @receiver(post_save, sender=Document)
def document_changed(sender, instance, **kwargs): def document_changed(sender, instance, **kwargs):
async('tasks.index_object', sender, instance, save=False) async_task('tasks.index_object', sender, instance, save=False)
# turn off result saving to not flood your database # turn off result saving to not flood your database
.. code-block:: python .. code-block:: python
@@ -177,7 +177,7 @@ here's an example of how you can have Django Q take care of your indexes in real
index.update_object(instance, using=backend) index.update_object(instance, using=backend)
Now every time a Document is saved, your indexes will be updated without causing a delay in your save action. Now every time a Document is saved, your indexes will be updated without causing a delay in your save action.
You could expand this to dealing with deletes, by adding a ``post_delete`` signal and calling ``index.remove_object`` in the async function. You could expand this to dealing with deletes, by adding a ``post_delete`` signal and calling ``index.remove_object`` in the async_task function.
.. _shell: .. _shell:
@@ -187,13 +187,13 @@ You can execute or schedule shell commands using Pythons :mod:`subprocess` modul
.. code-block:: python .. code-block:: python
from django_q.tasks import async, result from django_q.tasks import async_task, result
# make a backup copy of setup.py # make a backup copy of setup.py
async('subprocess.call', ['cp', 'setup.py', 'setup.py.bak']) async_task('subprocess.call', ['cp', 'setup.py', 'setup.py.bak'])
# call ls -l and dump the output # call ls -l and dump the output
task_id=async('subprocess.check_output', ['ls', '-l']) task_id=async_task('subprocess.check_output', ['ls', '-l'])
# get the result # get the result
dir_list = result(task_id) dir_list = result(task_id)
@@ -202,10 +202,10 @@ In Python 3.5 the subprocess module has changed quite a bit and returns a :class
.. code-block:: python .. code-block:: python
from django_q.tasks import async, result from django_q.tasks import async_task, result
# make a backup copy of setup.py # make a backup copy of setup.py
tid = async('subprocess.run', ['cp', 'setup.py', 'setup.py.bak']) tid = async_task('subprocess.run', ['cp', 'setup.py', 'setup.py.bak'])
# get the result # get the result
r=result(tid, 500) r=result(tid, 500)
@@ -220,22 +220,22 @@ In Python 3.5 the subprocess module has changed quite a bit and returns a :class
from subprocess import PIPE from subprocess import PIPE
# call ls -l and pipe the output # call ls -l and pipe the output
tid = async('subprocess.run', ['ls', '-l'], stdout=PIPE) tid = async_task('subprocess.run', ['ls', '-l'], stdout=PIPE)
# get the result # get the result
res = result(tid, 500) res = result(tid, 500)
# print the output # print the output
print(res.stdout) print(res.stdout)
Instead of :func:`async` you can of course also use :func:`schedule` to schedule commands. Instead of :func:`async_task` you can of course also use :func:`schedule` to schedule commands.
For regular Django management commands, it is easier to call them directly: For regular Django management commands, it is easier to call them directly:
.. code-block:: python .. code-block:: python
from django_q.tasks import async, schedule from django_q.tasks import async_task, schedule
async('django.core.management.call_command','clearsessions') async_task('django.core.management.call_command','clearsessions')
# or clear those sessions every hour # or clear those sessions every hour
@@ -255,7 +255,7 @@ Adapted from `Sebastian Raschka's blog <http://sebastianraschka.com/Articles/201
# Group example with Parzen-window estimation # Group example with Parzen-window estimation
import numpy import numpy
from django_q.tasks import async, result_group, delete_group from django_q.tasks import async_task, result_group, delete_group
# the estimation function # the estimation function
def parzen_estimation(x_samples, point_x, h): def parzen_estimation(x_samples, point_x, h):
@@ -279,9 +279,9 @@ Adapted from `Sebastian Raschka's blog <http://sebastianraschka.com/Articles/201
multivariate_normal(mu_vec, cov_mat, 10000) multivariate_normal(mu_vec, cov_mat, 10000)
widths = numpy.linspace(1.0, 1.2, 100) widths = numpy.linspace(1.0, 1.2, 100)
x = numpy.array([[0], [0]]) x = numpy.array([[0], [0]])
# async them with a group label to the cache backend # async_task them with a group label to the cache backend
for w in widths: for w in widths:
async(parzen_estimation, sample, x, w, async_task(parzen_estimation, sample, x, w,
group='parzen', cached=True) group='parzen', cached=True)
# return after 100 results # return after 100 results
return result_group('parzen', count=100, cached=True) return result_group('parzen', count=100, cached=True)
@@ -302,7 +302,7 @@ Alternatively the ``parzen_async()`` function can also be written with :func:`as
multivariate_normal(mu_vec, cov_mat, 10000) multivariate_normal(mu_vec, cov_mat, 10000)
widths = numpy.linspace(1.0, 1.2, 100) widths = numpy.linspace(1.0, 1.2, 100)
x = numpy.array([[0], [0]]) x = numpy.array([[0], [0]])
# async them with async iterable # async_task them with async_task iterable
args = [(sample, x, w) for w in widths] args = [(sample, x, w) for w in widths]
result_id = async_iter(parzen_estimation, args, cached=True) result_id = async_iter(parzen_estimation, args, cached=True)
# return the cached result or timeout after 10 seconds # return the cached result or timeout after 10 seconds
docs/group.rst
+6 -6
View File
@@ -2,15 +2,15 @@
Groups Groups
====== ======
You can group together results by passing :func:`async` the optional ``group`` keyword: You can group together results by passing :func:`async_task` the optional ``group`` keyword:
.. code-block:: python .. code-block:: python
# result group example # result group example
from django_q.tasks import async, result_group from django_q.tasks import async_task, result_group
for i in range(4): for i in range(4):
async('math.modf', i, group='modf') async_task('math.modf', i, group='modf')
# wait until the group has 4 results # wait until the group has 4 results
result = result_group('modf', count=4) result = result_group('modf', count=4)
@@ -66,14 +66,14 @@ You can also access group functions from a task result instance:
task.group_delete() task.group_delete()
print('Deleted group {}'.format(task.group)) print('Deleted group {}'.format(task.group))
or call them directly on :class:`Async` object: or call them directly on :class:`AsyncTask` object:
.. code-block:: python .. code-block:: python
from django_q.tasks import Async from django_q.tasks import AsyncTask
# add a task to the math group and run it cached # add a task to the math group and run it cached
a = Async('math.floor', 2.5, group='math', cached=True) a = AsyncTask('math.floor', 2.5, group='math', cached=True)
# wait until this tasks group has 10 results # wait until this tasks group has 10 results
result = a.result_group(count=10) result = a.result_group(count=10)
docs/index.rst
+1 -1
View File
@@ -24,7 +24,7 @@ Features
- Rollbar and Sentry support - Rollbar and Sentry support
Django Q is tested with: Python 2.7 & 3.6. Django 1.8.19 LTS, 1.11.11 and 2.0.x Django Q is tested with: Python 3.6. 3.7 Django 1.11.11 LTS and 2.0.x
Contents: Contents:
docs/install.rst
+9 -7
View File
@@ -27,12 +27,13 @@ Installation
Requirements Requirements
------------ ------------
Django Q is tested for Python 2.7 and 3.6 Django Q is tested for Python 3.6 and 3.7
- `Django <https://www.djangoproject.com>`__ - `Django <https://www.djangoproject.com>`__
Django Q aims to use as much of Django's standard offerings as possible Django Q aims to use as much of Django's standard offerings as possible
The code is tested against Django versions `1.8.19 LTS`, `1.11.11` and `2.0.x`. The code is tested against Django versions `1.11.11 LTS` and `2.0.x`.
Please note that Django versions below 2.0 do not support Python 3.7
- `Django-picklefield <https://github.com/gintas/django-picklefield>`__ - `Django-picklefield <https://github.com/gintas/django-picklefield>`__
@@ -122,14 +123,14 @@ Other known issues are:
Python Python
~~~~~~ ~~~~~~
The code is always tested against the latest version of Python 2 and Python 3 and we try to stay compatible with the last two versions of each. The code is always tested against the latest version Python 3 and we try to stay compatible with the last two versions of each.
Current tests are performed with Python 2.7.14 and 3.6.3 Current tests are performed with 3.6 and 3.7
If you do encounter any regressions with earlier versions, please submit an issue on `github <https://github.com/Koed00/django-q>`__ If you do encounter any regressions with earlier versions, please submit an issue on `github <https://github.com/Koed00/django-q>`__
.. note:: .. note::
Django 1.7.10 or earlier is not compatible with Python 3.5
Django releases before 1.11 are not officially supported on Python 3.6 Django releases before 1.11 are not officially supported on Python 3.6
Django releases before 2.0 are not supported on Python 3.7
Open-source packages Open-source packages
~~~~~~~~~~~~~~~~~~~~ ~~~~~~~~~~~~~~~~~~~~
@@ -139,9 +140,10 @@ You can reference the `requirements <https://github.com/Koed00/django-q/blob/mas
Django Django
~~~~~~ ~~~~~~
We strive to be compatible with last two major version of Django. We strive to be compatible with last two major version of Django.
At the moment this means we support the 1.8.19 LTS, 1.11.11 and 2.0.x releases. At the moment this means we support the 1.11.11 and 2.0.x releases.
You might find that Django Q still works fine with Django 1.7,1.9 and 1.10, but new releases are no longer tested for it. Since we are now no longer supporting Python 2, we can also not support older versions of Django that do not support Python 3.
For this you can always use the pre 1.0 release, but it's no longer maintained.
docs/iterable.rst
+2 -2
View File
@@ -10,7 +10,7 @@ If you have an iterable object with arguments for a function, you can use :func:
# set up a list of arguments for math.floor # set up a list of arguments for math.floor
iter = [i for i in range(100)] iter = [i for i in range(100)]
# async iter them # async_task iter them
id=async_iter('math.floor',iter) id=async_iter('math.floor',iter)
# wait for the collated result for 1 second # wait for the collated result for 1 second
@@ -48,7 +48,7 @@ Reference
.. py:function:: async_iter(func, args_iter,**kwargs) .. py:function:: async_iter(func, args_iter,**kwargs)
Runs iterable arguments against the cache backend and returns a single collated result. Runs iterable arguments against the cache backend and returns a single collated result.
Accepts the same options as :func:`async` except ``hook``. See also the :class:`Iter` class. Accepts the same options as :func:`async_task` except ``hook``. See also the :class:`Iter` class.
:param object func: The task function to execute :param object func: The task function to execute
:param args: An iterable containing arguments for the task function :param args: An iterable containing arguments for the task function
docs/schedules.rst
+2 -2
View File
@@ -27,7 +27,7 @@ You can manage them through the :ref:`admin_page` or directly from your code wit
schedule_type=Schedule.DAILY schedule_type=Schedule.DAILY
) )
# In case you want to use async options # In case you want to use q_options
schedule('math.sqrt', schedule('math.sqrt',
9, 9,
hook='hooks.print_result', hook='hooks.print_result',
@@ -103,7 +103,7 @@ Reference
:param int minutes: Number of minutes for the Minutes type. :param int minutes: Number of minutes for the Minutes type.
:param int repeats: Number of times to repeat schedule. -1=Always, 0=Never, n =n. :param int repeats: Number of times to repeat schedule. -1=Always, 0=Never, n =n.
:param datetime next_run: Next or first scheduled execution datetime. :param datetime next_run: Next or first scheduled execution datetime.
:param dict q_options: async options to use for this schedule :param dict q_options: options passed to async_task for this schedule
:param kwargs: optional keyword arguments for the scheduled function. :param kwargs: optional keyword arguments for the scheduled function.
.. class:: Schedule .. class:: Schedule
docs/tasks.rst
+35 -35
View File
@@ -4,22 +4,22 @@ Tasks
.. _async: .. _async:
async() async_task()
------- ------------
Use :func:`async` from your code to quickly offload tasks to the :class:`Cluster`: Use :func:`async_task` from your code to quickly offload tasks to the :class:`Cluster`:
.. code:: python .. code:: python
from django_q.tasks import async, result from django_q.tasks import async_task, result
# create the task # create the task
async('math.copysign', 2, -2) async_task('math.copysign', 2, -2)
# or with import and storing the id # or with import and storing the id
import math.copysign import math.copysign
task_id = async(copysign, 2, -2) task_id = async_task(copysign, 2, -2)
# get the result # get the result
task_result = result(task_id) task_result = result(task_id)
@@ -30,13 +30,13 @@ Use :func:`async` from your code to quickly offload tasks to the :class:`Cluster
# but in most cases you will want to use a hook: # but in most cases you will want to use a hook:
async('math.modf', 2.5, hook='hooks.print_result') async_task('math.modf', 2.5, hook='hooks.print_result')
# hooks.py # hooks.py
def print_result(task): def print_result(task):
print(task.result) print(task.result)
:func:`async` can take the following optional keyword arguments: :func:`async_task` can take the following optional keyword arguments:
hook hook
"""" """"
@@ -90,7 +90,7 @@ a single keyword dict named ``q_options``. This enables you to use these keyword
'group': 'math', 'group': 'math',
'timeout': 30} 'timeout': 30}
async('math.modf', 2.5, q_options=opts) async_task('math.modf', 2.5, q_options=opts)
Please note that this will override any other option keywords. Please note that this will override any other option keywords.
@@ -99,18 +99,18 @@ Please note that this will override any other option keywords.
or you need to configure Django Q to run in synchronous mode for testing using the :ref:`sync` option. or you need to configure Django Q to run in synchronous mode for testing using the :ref:`sync` option.
Async AsyncTask
----- ---------
Optionally you can use the :class:`Async` class to instantiate a task and keep everything in a single object.: Optionally you can use the :class:`AsyncTask` class to instantiate a task and keep everything in a single object.:
.. code-block:: python .. code-block:: python
# Async class instance example # AsyncTask class instance example
from django_q.tasks import Async from django_q.tasks import AsyncTask
# instantiate an async task # instantiate an async task
a = Async('math.floor', 1.5, group='math') a = AsyncTask('math.floor', 1.5, group='math')
# you can set or change keywords afterwards # you can set or change keywords afterwards
a.cached = True a.cached = True
@@ -136,7 +136,7 @@ Optionally you can use the :class:`Async` class to instantiate a task and keep e
1 1
2 2
Once you change any of the parameters of the task after it has run, the result is invalidated and you will have to :func:`Async.run` it again to retrieve a new result. Once you change any of the parameters of the task after it has run, the result is invalidated and you will have to :func:`AsyncTask.run` it again to retrieve a new result.
Cached operations Cached operations
----------------- -----------------
@@ -150,10 +150,10 @@ You can also opt to set a manual timeout on the results, by setting e.g. ``cache
This works both globally or on individual async executions.:: This works both globally or on individual async executions.::
# simple cached example # simple cached example
from django_q.tasks import async, result from django_q.tasks import async_task, result
# cache the result for 10 seconds # cache the result for 10 seconds
id = async('math.floor', 100, cached=10) id = async_task('math.floor', 100, cached=10)
# wait max 50ms for the result to appear in the cache # wait max 50ms for the result to appear in the cache
result(id, wait=50, cached=True) result(id, wait=50, cached=True)
@@ -169,15 +169,15 @@ As you can see you can easily turn a cached result into a permanent database res
This also works for group actions:: This also works for group actions::
# cached group example # cached group example
from django_q.tasks import async, result_group from django_q.tasks import async_task, result_group
from django_q.brokers import get_broker from django_q.brokers import get_broker
# set up a broker instance for better performance # set up a broker instance for better performance
broker = get_broker() broker = get_broker()
# async a hundred functions under a group label # Async a hundred functions under a group label
for i in range(100): for i in range(100):
async('math.frexp', async_task('math.frexp',
i, i,
group='frexp', group='frexp',
cached=True, cached=True,
@@ -191,13 +191,13 @@ If you don't need hooks, that exact same result can be achieved by using the mor
Synchronous testing Synchronous testing
------------------- -------------------
:func:`async` can be instructed to execute a task immediately by setting the optional keyword ``sync=True``. :func:`async_task` can be instructed to execute a task immediately by setting the optional keyword ``sync=True``.
The task will then be injected straight into a worker and the result saved by a monitor instance:: The task will then be injected straight into a worker and the result saved by a monitor instance::
from django_q.tasks import async, fetch from django_q.tasks import async_task, fetch
# create a synchronous task # create a synchronous task
task_id = async('my.buggy.code', sync=True) task_id = async_task('my.buggy.code', sync=True)
# the task will then be available immediately # the task will then be available immediately
task = fetch(task_id) task = fetch(task_id)
@@ -210,24 +210,24 @@ The task will then be injected straight into a worker and the result saved by a
An error occurred: ImportError("No module named 'my'",) An error occurred: ImportError("No module named 'my'",)
Note that :func:`async` will block until the task is executed and saved. This feature bypasses the broker and is intended for debugging and development. Note that :func:`async_task` will block until the task is executed and saved. This feature bypasses the broker and is intended for debugging and development.
Instead of setting ``sync`` on each individual ``async`` you can also configure :ref:`sync` as a global override. Instead of setting ``sync`` on each individual ``async_task`` you can also configure :ref:`sync` as a global override.
Connection pooling Connection pooling
------------------ ------------------
Django Q tries to pass broker instances around its parts as much as possible to save you from running out of connections. Django Q tries to pass broker instances around its parts as much as possible to save you from running out of connections.
When you are making individual calls to :func:`async` a lot though, it can help to set up a broker to reuse for :func:`async`: When you are making individual calls to :func:`async_task` a lot though, it can help to set up a broker to reuse for :func:`async_task`:
.. code:: python .. code:: python
# broker connection economy example # broker connection economy example
from django_q.tasks import async from django_q.tasks import async_task
from django_q.brokers import get_broker from django_q.brokers import get_broker
broker = get_broker() broker = get_broker()
for i in range(50): for i in range(50):
async('math.modf', 2.5, broker=broker) async_task('math.modf', 2.5, broker=broker)
.. tip:: .. tip::
@@ -237,7 +237,7 @@ When you are making individual calls to :func:`async` a lot though, it can help
Reference Reference
--------- ---------
.. py:function:: async(func, *args, hook=None, group=None, timeout=None,\ .. py:function:: async_task(func, *args, hook=None, group=None, timeout=None,\
save=None, sync=False, cached=False, broker=None, q_options=None, **kwargs) save=None, sync=False, cached=False, broker=None, q_options=None, **kwargs)
Puts a task in the cluster queue Puts a task in the cluster queue
@@ -249,7 +249,7 @@ Reference
:param int timeout: Overrides global cluster :ref:`timeout`. :param int timeout: Overrides global cluster :ref:`timeout`.
:param bool save: Overrides global save setting for this task. :param bool save: Overrides global save setting for this task.
:param bool ack_failure: Overrides the global :ref:`ack_failures` setting for this task. :param bool ack_failure: Overrides the global :ref:`ack_failures` setting for this task.
:param bool sync: If set to True, async will simulate a task execution :param bool sync: If set to True, async_task will simulate a task execution
:param cached: Output the result to the cache backend. Bool or timeout in seconds :param cached: Output the result to the cache backend. Bool or timeout in seconds
:param broker: Optional broker connection from :func:`brokers.get_broker` :param broker: Optional broker connection from :func:`brokers.get_broker`
:param dict q_options: Options dict, overrides option keywords :param dict q_options: Options dict, overrides option keywords
@@ -408,13 +408,13 @@ Reference
A proxy model of :class:`Task` with the queryset filtered on :attr:`Task.success` is ``False``. A proxy model of :class:`Task` with the queryset filtered on :attr:`Task.success` is ``False``.
.. py:class:: Async(func, *args, **kwargs) .. py:class:: AsyncTask(func, *args, **kwargs)
A class wrapper for the :func:`async` function. A class wrapper for the :func:`async_task` function.
:param object func: The task function to execute :param object func: The task function to execute
:param tuple args: The arguments for the task function :param tuple args: The arguments for the task function
:param dict kwargs: Keyword arguments for the task function, including async options :param dict kwargs: Keyword arguments for the task function, including async_task options
.. py:attribute:: id .. py:attribute:: id
@@ -434,7 +434,7 @@ Reference
.. py:attribute:: kwargs .. py:attribute:: kwargs
Keyword arguments for the function. Can include any of the optional async keyword attributes directly or in a `q_options` dictionary. Keyword arguments for the function. Can include any of the optional async_task keyword attributes directly or in a `q_options` dictionary.
.. py:attribute:: broker .. py:attribute:: broker
requirements.in
-1
View File
@@ -8,4 +8,3 @@ django-redis
iron-mq iron-mq
boto3 boto3
pymongo pymongo
rollbar
requirements.txt
+13 -14
View File
@@ -5,28 +5,27 @@
# pip-compile --output-file requirements.txt requirements.in # pip-compile --output-file requirements.txt requirements.in
# #
arrow==0.12.1 arrow==0.12.1
blessed==1.14.2 blessed==1.15.0
boto3==1.6.7 boto3==1.7.76
botocore==1.9.7 # via boto3, s3transfer botocore==1.10.76 # via boto3, s3transfer
certifi==2018.1.18 # via requests certifi==2018.8.13 # via requests
chardet==3.0.4 # via requests chardet==3.0.4 # via requests
django-picklefield==1.0.0 django-picklefield==1.0.0
django-redis==4.9.0 django-redis==4.9.0
django==2.0 # via django-redis django==2.1 # via django-redis
docutils==0.14 # via botocore docutils==0.14 # via botocore
hiredis==0.2.0 hiredis==0.2.0
idna==2.6 # via requests idna==2.7 # via requests
iron-core==1.2.0 # via iron-mq iron-core==1.2.0 # via iron-mq
iron-mq==0.9 iron-mq==0.9
jmespath==0.9.3 # via boto3, botocore jmespath==0.9.3 # via boto3, botocore
psutil==5.4.3 psutil==5.4.6
pymongo==3.6.1 pymongo==3.7.1
python-dateutil==2.7.0 # via arrow, botocore, iron-core python-dateutil==2.7.3 # via arrow, botocore, iron-core
pytz==2018.3 # via django pytz==2018.5 # via django
redis==2.10.6 redis==2.10.6
requests==2.18.4 # via iron-core, rollbar requests==2.19.1 # via iron-core
rollbar==0.13.18
s3transfer==0.1.13 # via boto3 s3transfer==0.1.13 # via boto3
six==1.11.0 # via blessed, python-dateutil, rollbar six==1.11.0 # via blessed, python-dateutil
urllib3==1.22 # via requests urllib3==1.23 # via requests
wcwidth==0.1.7 # via blessed wcwidth==0.1.7 # via blessed
setup.py
+6 -6
View File
@@ -1,10 +1,10 @@
import os import os
from setuptools import setup, Command
with open(os.path.join(os.path.dirname(__file__), 'README.rst')) as readme: with open(os.path.join(os.path.dirname(__file__), 'README.rst')) as readme:
README = readme.read() README = readme.read()
os.chdir(os.path.normpath(os.path.join(os.path.abspath(__file__), os.pardir))) os.chdir(os.path.normpath(os.path.join(os.path.abspath(__file__), os.pardir)))
from setuptools import setup, Command
class PyTest(Command): class PyTest(Command):
@@ -26,7 +26,7 @@ class PyTest(Command):
setup( setup(
name='django-q', name='django-q',
version='0.9.4', version='1.0.0',
author='Ilan Steemers', author='Ilan Steemers',
author_email='koed0@gmail.com', author_email='koed0@gmail.com',
keywords='django distributed task queue worker scheduler cron redis disque ironmq sqs orm mongodb multiprocessing rollbar', keywords='django distributed task queue worker scheduler cron redis disque ironmq sqs orm mongodb multiprocessing rollbar',
@@ -36,11 +36,11 @@ setup(
license='MIT', license='MIT',
description='A multiprocessing distributed task queue for Django', description='A multiprocessing distributed task queue for Django',
long_description=README, long_description=README,
install_requires=['django>=1.8', 'django-picklefield', 'blessed', 'arrow'], install_requires=['django>=1.11', 'django-picklefield', 'blessed', 'arrow'],
test_requires=['pytest', 'pytest-django', ], test_requires=['pytest', 'pytest-django', ],
cmdclass={'test': PyTest}, cmdclass={'test': PyTest},
classifiers=[ classifiers=[
'Development Status :: 4 - Beta', 'Development Status :: 5 - Production/Stable',
'Environment :: Web Environment', 'Environment :: Web Environment',
'Framework :: Django', 'Framework :: Django',
'Intended Audience :: Developers', 'Intended Audience :: Developers',
@@ -48,13 +48,13 @@ setup(
'Operating System :: POSIX', 'Operating System :: POSIX',
'Operating System :: MacOS', 'Operating System :: MacOS',
'Programming Language :: Python', 'Programming Language :: Python',
'Programming Language :: Python :: 2',
'Programming Language :: Python :: 2.7',
'Programming Language :: Python :: 3', 'Programming Language :: Python :: 3',
'Programming Language :: Python :: 3.4', 'Programming Language :: Python :: 3.4',
'Programming Language :: Python :: 3.5', 'Programming Language :: Python :: 3.5',
'Programming Language :: Python :: 3.6', 'Programming Language :: Python :: 3.6',
'Programming Language :: Python :: 3.7',
'Topic :: Internet :: WWW/HTTP', 'Topic :: Internet :: WWW/HTTP',
'Topic :: System :: Distributed Computing',
'Topic :: Software Development :: Libraries :: Python Modules', 'Topic :: Software Development :: Libraries :: Python Modules',
], ],
entry_points={ entry_points={