如何从Django管理员触发Celery任务? [英] How do I trigger a Celery task from Django admin?

问题描述

我有一个名为 Publication 的模型，我想在Django Admin的列表视图中添加一个按钮，该按钮将允许触发Celery任务.

I have a model called Publication and I'd like to add a button to the list view in Django Admin which would allow triggering a Celery task.

admin.py :

from django.contrib import admin
from .models import Publication


class PublicationAdmin(admin.ModelAdmin):
    change_list_template = "variants/admin_publication_list.html"

    def update(self, request):
        # trigger task
        # redirect to admin list

admin.site.register(Publication, PublicationAdmin)

variants/admin_publication_list.html :

{% extends 'admin/change_list.html' %}

{% block object-tools %}
  <li>
    <a href="/admin/variants/publication/update/">
      Update
    </a>
  </li>
  {{ block.super }}
{% endblock %}

但是，当我按下按钮时，我只会收到通知:

However when I press a button I only get a notice:

ID为"update"的出版物不存在.也许它被删除了?

Publication with ID "update" doesn’t exist. Perhaps it was deleted?

推荐答案

方法名称 update 或路由名称 update 太通用，有时框架会自动使用，请尝试命名它与您的功能相匹配.撤销出版物或 sync-publications .

Method name update or route name update is too generic and sometimes used by frameworks automatically, try naming it to match your functionality. revoke-publications or sync-publications.

admin.py :

from django.contrib import admin
from django.urls import path
from django.http import HttpResponseRedirect
from .models import Publication
from .tasks import your_celery_task

@admin.register(Publication)
class PublicationAdmin(admin.ModelAdmin):
    change_list_template = "variants/admin_publication_list.html"

    def get_urls(self):
        urls = super().get_urls()
        my_urls = [
            path('update-publications/', self.publication_update),
        ]
        return my_urls + urls

    def publication_update(self, request):
          result = your_celery_task.delay("some_arg")
          self.message_user(
              request,
              f"Task with the ID: {result.task_id} was added to queue. Last known status: {result.status}"
          )
          return HttpResponseRedirect("../")

variants/admin_publication_list.html :

{% extends 'admin/change_list.html' %}

{% block object-tools %}
    <div>
        <form action="update-publications/" method="POST">
            {% csrf_token %}
                <button type="submit">Run Celery Task</button>
        </form>
    </div>
    <br />
    {{ block.super }}
{% endblock %}

这篇关于如何从Django管理员触发Celery任务?的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！