Merge pull request #466 from timomeara/master

Add attempt_count to limit the number of times a filed task will be re-attempted
This commit is contained in:
Ilan Steemers
2020-08-15 10:21:33 +02:00
committed by GitHub
7 changed files with 95 additions and 0 deletions
+6
View File
@@ -478,7 +478,12 @@ def save_task(task, broker: Broker):
existing_task.stopped = task["stopped"] existing_task.stopped = task["stopped"]
existing_task.result = task["result"] existing_task.result = task["result"]
existing_task.success = task["success"] existing_task.success = task["success"]
existing_task.attempt_count = existing_task.attempt_count + 1
existing_task.save() existing_task.save()
if Conf.MAX_ATTEMPTS > 0 and existing_task.attempt_count >= Conf.MAX_ATTEMPTS:
broker.acknowledge(task['ack_id'])
else: else:
Task.objects.create( Task.objects.create(
id=task["id"], id=task["id"],
@@ -492,6 +497,7 @@ def save_task(task, broker: Broker):
result=task["result"], result=task["result"],
group=task.get("group"), group=task.get("group"),
success=task["success"], success=task["success"],
attempt_count=1
) )
except Exception as e: except Exception as e:
logger.error(e) logger.error(e)
+3
View File
@@ -164,6 +164,9 @@ class Conf:
# Optional error reporting setup # Optional error reporting setup
ERROR_REPORTER = conf.get("error_reporter", {}) ERROR_REPORTER = conf.get("error_reporter", {})
# Optional attempt count. set to 0 for infinite attempts
MAX_ATTEMPTS = conf.get('max_attempts', 0)
# OSX doesn't implement qsize because of missing sem_getvalue() # OSX doesn't implement qsize because of missing sem_getvalue()
try: try:
QSIZE = Queue().qsize() == 0 QSIZE = Queue().qsize() == 0
@@ -0,0 +1,18 @@
# Generated by Django 3.0.7 on 2020-08-11 15:17
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
('django_q', '0012_auto_20200702_1608'),
]
operations = [
migrations.AddField(
model_name='task',
name='attempt_count',
field=models.IntegerField(default=0),
),
]
+1
View File
@@ -29,6 +29,7 @@ class Task(models.Model):
started = models.DateTimeField(editable=False) started = models.DateTimeField(editable=False)
stopped = models.DateTimeField(editable=False) stopped = models.DateTimeField(editable=False)
success = models.BooleanField(default=True, editable=False) success = models.BooleanField(default=True, editable=False)
attempt_count = models.IntegerField(default=0)
@staticmethod @staticmethod
def get_result(task_id): def get_result(task_id):
+35
View File
@@ -397,6 +397,41 @@ def test_bad_secret(broker, monkeypatch):
broker.delete_queue() broker.delete_queue()
@pytest.mark.django_db
def test_attempt_count(broker, monkeypatch):
monkeypatch.setattr(Conf, 'MAX_ATTEMPTS', 3)
tag = uuid()
task = {'id': tag[1],
'name': tag[0],
'func': 'math.copysign',
'args': (1, -1),
'kwargs': {},
'started': timezone.now(),
'stopped': timezone.now(),
'success': False,
'result': None}
# initial save - no success
save_task(task, broker)
assert Task.objects.filter(id=task['id']).exists()
saved_task = Task.objects.get(id=task['id'])
assert saved_task.attempt_count == 1
sleep(0.5)
# second save
old_stopped = task['stopped']
task['stopped'] = timezone.now()
save_task(task, broker)
saved_task = Task.objects.get(id=task['id'])
assert saved_task.attempt_count == 2
# third save -
task['stopped'] = timezone.now()
save_task(task, broker)
saved_task = Task.objects.get(id=task['id'])
assert saved_task.attempt_count == 3
# task should be removed from queue
assert broker.queue_size() == 0
@pytest.mark.django_db @pytest.mark.django_db
def test_update_failed(broker): def test_update_failed(broker):
tag = uuid() tag = uuid()
+23
View File
@@ -33,6 +33,29 @@ You can resubmit a failed task back to the queue using the admins action menu.
Uses the :class:`Failure` proxy model Uses the :class:`Failure` proxy model
Customize the admin UI by creating your own ``admin.ModelAdmin`` class and use ``admin.site.unregister`` and ``admin.site.register`` to replace the default
for example:
.. code-block:: python
from django_q import models as q_models
from django_q import admin as q_admin
admin.site.unregister([q_models.Failure])
@admin.register(q_models.Failure)
class ChildClassAdmin(q_admin.FailAdmin):
list_display = (
'name',
'func',
'result',
'started',
# add attempt_count to list_display
'attempt_count'
)
Scheduled tasks Scheduled tasks
--------------- ---------------
+9
View File
@@ -75,6 +75,15 @@ 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_task`. 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``.
.. _max_attempts:
max_attempts
~~~~~~~~~~~~~
Limit the number of retry attempts for failed tasks. Set to 0 for infinite retries. Defaults to 0
.. _retry: .. _retry:
retry retry