如何在Django中定期更新后台任务状态？

1 人关注

我有一个django项目，其中包含一个长期运行的进程。我为此使用了django-background-tasks库。它是有效的，但我想为用户创建一个待处理页面，并显示任务的状态。我应该每60秒刷新一次页面并更新状态。我怎样才能做到这一点呢？

python

django

background-process

background-task

long-running-processes

Aslı Kök

发布于 2020-03-10

2 个回答

Balasundar

发布于 2020-03-10

已采纳

0 人赞同

希望你了解阿贾克斯。

如何用Django使用Ajax。 https://simpleisbetterthancomplex.com/tutorial/2016/08/29/how-to-work-with-ajax-request-with-django.html

如何每隔N秒运行一次Ajax代码。 https://thisinterestsme.com/ajax-request-every-10-seconds/

如果你想部分加载一个页面，那么你应该将该页面分成两部分。一部分应该包含你想刷新或重新加载的特定div，将页面名称定为partial_load.html。而另一个页面可以有其余的代码，考虑文件名为full_page.html。通过使用include标签（{% include "partial_load.html" %}），可以将partial_load.html包含在full_page.html中。

 def refresh_page(request):
      if request.is_ajax():
           now = timezone.now()
           pending, running = get_process_status()
           context = {
             "pending": count_pending, 
             "running": count_running
           return render(request, "partial_load.html", context)
full_page.html
  <div> Something </div>
   .....
   <div id="status">
    {% include "partial_load.html" %}
  ......
  <div> some more stuff </div>
 </html>


         0
         
         人赞同


          
           如果你在安装background_task包后迁移了数据库的变化（那么只有数据库有background_task包的表）。
你可以通过简单地查询background_task模型来获得在后台运行的进程的状态，就像查询其他用户定义的模型一样。
          
          from background_task.models import Task, CompletedTask
from django.utils import timezone
def get_process_status(parameters):
     now = timezone.now()
     # pending tasks will have `run_at` column greater than current time.
     # Similar for running tasks, it shall be
     # greater than or equal to `locked_at` column.
     # Running tasks won't work with SQLite DB,
     # because of concurrency issues in SQLite.
     # If your task still not started running and waiting in the queue, then you  can find it in pending_tasks
     pending_tasks = Task.objects.filter(run_at__gt=now)
     # If your your task is in running state, you can find it in running_tasks
     running_tasks = Task.objects.filter(locked_at__gte=now)
     # Completed tasks goes in `CompletedTask` model.
     # I have picked all, you can choose to filter based on what you want.
     # If your task completed you can find it in Completed task model.
     completed_tasks = CompletedTask.objects.all()
     # If you want the result in json format just add .values() at the end of the
     # ORM query like "completed_tasks = CompletedTask.objects.all().values()"     
     print(completed_tasks, running_tasks, pending_tasks)
     ......
     ......
     return process_status
如果你想每60秒运行一次该函数，请使用background_task安排任务。
示例代码。
@background(schedule=60)
def get_process_status(parameters):
  .....