diff --git a/README.md b/README.md index d88e20d..73d0d1d 100644 --- a/README.md +++ b/README.md @@ -16,8 +16,41 @@ Schedule the asynchronous execution of a function by calling `async` from within ```python async(func,*args,hook=None,**kwargs) ``` -The optional hook function gets the finished task object as its first argument after execution +####Basic example +```python +from django_q import async +# math.copysign(2,-2) +async('math.copysign', 2, -2) + +# also +from math import copysign + +async(copysign, 2, -2) + +``` +#### Result example +```python +from django_q import async, result + +# create the task +task_id = async('math.copysign', 2, -2) + +# get the result +task_result = result(task_id) + +# result returns None if the task has not been executed yet +# so it makes more sense to use a hook: +from hooks import print_result + +async('math.modf', 2.5, hook='hooks.print_result') + +# hooks.py +def print_result(task): + print(task.result) + + +``` ### Management commands #### `qcluster` diff --git a/README.rst b/README.rst index 9c6d501..5636c95 100644 --- a/README.rst +++ b/README.rst @@ -25,8 +25,43 @@ from within your Django project. async(func,*args,hook=None,**kwargs) -The optional hook function gets the finished task object as its first -argument after execution +Basic example +^^^^^^^^^^^^^ + +.. code:: python + + from django_q import async + + # math.copysign(2,-2) + async('math.copysign', 2, -2) + + # also + from math import copysign + + async(copysign, 2, -2) + +Result example +^^^^^^^^^^^^^^ + +.. code:: python + + from django_q import async, result + + # create the task + task_id = async('math.copysign', 2, -2) + + # get the result + task_result = result(task_id) + + # result returns None if the task has not been executed yet + # so it makes more sense to use a hook: + from hooks import print_result + + async('math.modf', 2.5, hook='hooks.print_result') + + # hooks.py + def print_result(task): + print(task.result) Management commands ~~~~~~~~~~~~~~~~~~~ diff --git a/django_q/core.py b/django_q/core.py index 26d815b..aeabc31 100644 --- a/django_q/core.py +++ b/django_q/core.py @@ -132,7 +132,8 @@ class Cluster(object): class Sentinel(object): - def __init__(self, stop_event, start_event, list_key=Q_LIST): + def __init__(self, stop_event, start_event, list_key=Q_LIST, start=True): + # Make sure we catch signals for the pool signal.signal(signal.SIGINT, signal.SIG_IGN) signal.signal(signal.SIGTERM, signal.SIG_DFL) self.pid = current_process().pid @@ -151,8 +152,9 @@ class Sentinel(object): self.event_out = Event() self.monitor_pid = None self.pusher_pid = None - self.spawn_cluster() - self.guard() + if start: + self.spawn_cluster() + self.guard() def spawn_process(self, target, *args): # This is just for PyCharm to not crash. Ignore it. @@ -210,8 +212,9 @@ class Sentinel(object): Stat(self).save() if self.stop_event.is_set(): break + # Call scheduler once a minute (or so) counter += 1 - if counter > 15: + if counter > 30: counter = 0 scheduler() sleep(2) @@ -269,6 +272,14 @@ def monitor(done_queue): def worker(task_queue, done_queue): name = current_process().name logger.info('{} ready for work at {}'.format(name, current_process().pid)) + task = {} + + def return_pack(res, success): + task['result'] = res + task['stopped'] = timezone.now() + task['success'] = success + done_queue.put(task) + for pack in iter(task_queue.get, 'STOP'): # unpickle the task try: @@ -278,30 +289,33 @@ def worker(task_queue, done_queue): continue except signing.BadSignature as e: task['name'] = task['name'].rsplit(":", 1)[0] - task['stopped'] = timezone.now() - task['result'] = e - task['success'] = False - done_queue.put(task) + return_pack(e, False) continue - module, func = task['func'].rsplit('.', 1) logger.info('{} processing [{}]'.format(name, task['name'])) + f = task['func'] + # if it's not an instance try to get it from the string + if not callable(task['func']): + try: + module, func = f.rsplit('.', 1) + m = importlib.import_module(module) + f = getattr(m, func) + except (ValueError, ImportError, AttributeError) as e: + logger.error(e) + return_pack(e, False) + continue + # execute the payload try: - m = importlib.import_module(module) - f = getattr(m, func) - task['result'] = f(*task['args'], **task['kwargs']) - task['stopped'] = timezone.now() - task['success'] = True - done_queue.put(task) - gc.collect() + result = f(*task['args'], **task['kwargs']) + return_pack(result, True) except Exception as e: - task['result'] = e - task['stopped'] = timezone.now() - task['success'] = False - done_queue.put(task) + return_pack(e, False) logger.info('{} stopped doing work'.format(name)) def save_task(task): + """ + Saves the task package to Django + """ if task['success'] and 0 < SAVE_LIMIT < Success.objects.count(): Success.objects.first().delete() Task.objects.create(name=task['name'], @@ -451,19 +465,22 @@ class Stat(Status): def scheduler(): for schedule in Schedule.objects.exclude(repeats=0).filter(next_run__lt=timezone.now()): args = () - kwargs= {} + kwargs = {} + # get args, kwargs and hook if schedule.kwargs: try: + # eval should be safe here cause dict() kwargs = eval('dict({})'.format(schedule.kwargs)) except SyntaxError: - kwargs={} + kwargs = {} if schedule.args: args = ast.literal_eval(schedule.args) + # single value won't eval to tuple, so: if type(args) != tuple: args = (args,) if schedule.hook: kwargs['hook'] = schedule.hook - schedule.task = async(schedule.func, *args, **kwargs) + # set up the next run time if not schedule.schedule_type == schedule.ONCE: next_run = arrow.get(schedule.next_run) if schedule.schedule_type == schedule.HOURLY: @@ -482,6 +499,8 @@ def scheduler(): schedule.repeats += -1 else: schedule.repeats = 0 + # send it to the cluster + schedule.task = async(schedule.func, *args, **kwargs) if not schedule.task: logger.error('{} failed to create task from schedule {}').format(current_process().name, schedule.id) else: diff --git a/django_q/migrations/0001_initial.py b/django_q/migrations/0001_initial.py index 4e4287d..4ea2def 100644 --- a/django_q/migrations/0001_initial.py +++ b/django_q/migrations/0001_initial.py @@ -15,26 +15,26 @@ class Migration(migrations.Migration): migrations.CreateModel( name='Schedule', fields=[ - ('id', models.AutoField(serialize=False, primary_key=True, auto_created=True, verbose_name='ID')), - ('func', models.CharField(max_length=256)), - ('hook', models.CharField(blank=True, max_length=256, null=True)), - ('args', models.CharField(blank=True, max_length=256, null=True)), - ('kwargs', models.CharField(blank=True, max_length=256, null=True)), - ('schedule_type', models.CharField(choices=[('O', 'Once'), ('H', 'Hourly'), ('D', 'Daily'), ('W', 'Weekly'), ('M', 'Monthly'), ('Q', 'Quarterly'), ('Y', 'Yearly')], max_length=1, verbose_name='Schedule Type', default='O')), - ('repeats', models.SmallIntegerField(verbose_name='Repeats', default=-1)), - ('next_run', models.DateTimeField(null=True, verbose_name='Next Run', default=django.utils.timezone.now)), + ('id', models.AutoField(serialize=False, auto_created=True, primary_key=True, verbose_name='ID')), + ('func', models.CharField(help_text='e.g. module.tasks.function', max_length=256)), + ('hook', models.CharField(blank=True, help_text='e.g. module.tasks.result_function', max_length=256, null=True)), + ('args', models.CharField(blank=True, help_text="e.g. 1, 2, 'John'", max_length=256, null=True)), + ('kwargs', models.CharField(blank=True, help_text="e.g. x=1, y=2, name='John'", max_length=256, null=True)), + ('schedule_type', models.CharField(choices=[('O', 'Once'), ('H', 'Hourly'), ('D', 'Daily'), ('W', 'Weekly'), ('M', 'Monthly'), ('Q', 'Quarterly'), ('Y', 'Yearly')], default='O', verbose_name='Schedule Type', max_length=1)), + ('repeats', models.SmallIntegerField(help_text='n = n times, -1 = forever', default=-1, verbose_name='Repeats')), + ('next_run', models.DateTimeField(default=django.utils.timezone.now, null=True, verbose_name='Next Run')), ('task', models.CharField(editable=False, max_length=100, null=True)), ], options={ - 'ordering': ['next_run'], 'verbose_name': 'Scheduled task', + 'ordering': ['next_run'], }, ), migrations.CreateModel( name='Task', fields=[ - ('id', models.AutoField(serialize=False, primary_key=True, auto_created=True, verbose_name='ID')), - ('name', models.CharField(editable=False, max_length=100)), + ('id', models.AutoField(serialize=False, auto_created=True, primary_key=True, verbose_name='ID')), + ('name', models.CharField(max_length=100, editable=False)), ('func', models.CharField(max_length=256)), ('hook', models.CharField(max_length=256, null=True)), ('args', picklefield.fields.PickledObjectField(editable=False)), diff --git a/django_q/models.py b/django_q/models.py index d459b6a..59cefcc 100644 --- a/django_q/models.py +++ b/django_q/models.py @@ -37,14 +37,19 @@ class Task(models.Model): @receiver(pre_save, sender=Task) def call_hook(sender, instance, **kwargs): if instance.hook: - module, func = instance.hook.rsplit('.', 1) + logger = logging.getLogger('django-q') + f = instance.hook + if not callable(f): + try: + module, func = f.rsplit('.', 1) + m = importlib.import_module(module) + f = getattr(m, func) + except (ValueError, ImportError, AttributeError): + logger.error(_('malformed return hook \'{}\' for {}').format(instance.hook, instance.name)) try: - m = importlib.import_module(module) - f = getattr(m, func) f(instance) except Exception as e: - logger = logging.getLogger('django-q') - logger.error(_('return hook failed on {}').format(instance.name)) + logger.error(_('return hook {} failed on {}').format(instance.hook, instance.name)) logger.exception(e) diff --git a/django_q/tests/test_cluster.py b/django_q/tests/test_cluster.py index d8abd9e..2c6871b 100644 --- a/django_q/tests/test_cluster.py +++ b/django_q/tests/test_cluster.py @@ -11,6 +11,7 @@ sys.path.insert(0, myPath + '/../') from django_q.core import Cluster, r, async, pusher, worker, monitor, Sentinel from django_q.humanhash import DEFAULT_WORDLIST from django_q import result, get_task +from django_q.tests.tasks import multiply class WordClass(object): @@ -20,6 +21,7 @@ class WordClass(object): def get_words(self): return self.word_list + def test_redis_connection(): assert r.ping() is True @@ -43,6 +45,7 @@ def test_sentinel(): Sentinel(stop_event, start_event, list_key='sentinel_test:q') assert start_event.is_set() + @pytest.mark.django_db def test_cluster(): list_key = 'cluster_test:q' @@ -72,6 +75,7 @@ def test_cluster(): assert result(task) == 1506 r.delete(list_key) + @pytest.mark.django_db def run_cluster(): list_key = 'run_test:q' @@ -83,6 +87,7 @@ def run_cluster(): assert c.stop() is True r.delete(list_key) + @pytest.mark.django_db def blah_async(): a = async('django_q.tests.tasks.count_letters', DEFAULT_WORDLIST, hook='django_q.tests.test_q.assert_result') @@ -91,35 +96,50 @@ def blah_async(): c = async('django_q.tests.tasks.count_letters', DEFAULT_WORDLIST, 'oneargumentoomany', hook='django_q.tests.test_q.assert_bad_result') # unknown function - d = async('django_q.tests.tasks.does_not_exist', WordClass(), hook='django_q.tests.test_q.assert_bad_result') + d = async('django_q.tests.tasks.does_not_exist', WordClass(), hook='django_q.tests.test_q.assert_bad_result') # function without result e = async('django_q.tests.tasks.countdown', 100000) + # function as instance + f = async(multiply, 753, 2, hook=assert_result) + # check if everything has a task name assert isinstance(a, str) assert isinstance(b, str) assert isinstance(c, str) assert isinstance(d, str) assert isinstance(e, str) + assert isinstance(f, str) + # run the cluster to execute the tasks run_cluster() + # task a result_a = get_task(a) assert result_a is not None assert result_a.success is True assert result(a) == 1506 + # task b result_b = get_task(b) assert result_b is not None assert result_b.success is True assert result(b) == 1506 + # task c result_c = get_task(c) assert result_c is not None assert result_c.success is False + # task d result_d = get_task(d) assert result_d is not None assert result_d.success is False + # task e result_e = get_task(e) assert result_e is not None assert result_e.success is True - assert result(b) is None - + assert result(e) is None + # task f + result_f = get_task(f) + assert result_f is not None + assert result_f.success is True + assert result(f) == 1506 +# not sure if this actually asserts, but it is called @pytest.mark.django_db def assert_result(task): assert task is not None diff --git a/requirements.txt b/requirements.txt index d891364..722cee9 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,4 +1,4 @@ -arrow +arrow==0.5.4 blessed==1.9.5 coloredlogs==1.0.1 django-picklefield==0.3.1