Retry Task button in Admin panel

[vsmelov]

How can I retry to execute a task if it fails due to some temporary problem from the admin panel?

I would like just to send the new task with the same arguments, how can I do that?
Thank you.

[iarp]

I've wanted something like this for a long time now.

I unregister TaskResult from admin site and re-register with my own class extending the original TaskResultAdmin as seen below.

I place this code in my "core/admin.py" app, where I keep my User model and other user based things.

import ast
import json
import importlib


class CeleryTaskResultAdminOverride(TaskResultAdmin):

    actions = ('execute_task',)

    def execute_task(modeladmin, request, queryset):
        for task_result in queryset.all():  # type: TaskResult
            args = ast.literal_eval(json.loads(task_result.task_args))
            kwargs = ast.literal_eval(json.loads(task_result.task_kwargs))

            pkg, attr = task_result.task_name.rsplit(".", 1)
            task = getattr(importlib.import_module(pkg), attr)

            task.apply_async(args=args, kwargs=kwargs)

admin.site.unregister(TaskResult)
admin.site.register(TaskResult, CeleryTaskResultAdminOverride)