python - 如何使用 django-background-tasks

Question

我正在制作一个 Django 应用程序。为了根据行和评论计算提要的排名，我正在尝试使用 django-background-tasks。我在节点模型中使用的功能是:

    @background(schedule=60)
    def get_score(self):
        p = self.likes+self.comments    # popularity
        t = (now()-self.date).total_seconds()/3600  # age_in_hrs
        # last_activity =
        n = self.admin_score
        score = (p/pow((t+1), 1.2))*n
        self.score = score
        return score

但我没有看到分数有任何变化。这意味着我正在以正确的方式做这件事，但我缺少基本概念。有人可以告诉我如何使用 django-background-tasks 来安排任务或向我推荐一些现有的文档。

Answer 1

由于这个问题似乎很笼统，我相信这是基于我个人经验的关于“如何使用 django-background-tasks”的快速备忘单的正确位置。希望我不会是唯一一个使用它的人:)
环境

Python 3.8

Django 3.1

安装
我喜欢 pipenv所以:

> cd [my-django-project root directory]
> pipenv install django-background-tasks

现在将 'background_task' 添加到 settings.py 中的 INSTALLED_APPS:

INSTALLED_APPS = (
    # ...
    'background_task',
    # ...
)

并执行数据库迁移以确保 django-background-tasks 架构到位:

> pipenv shell
(my-django-project) bash-3.2$  python manage.py migrate

创建和注册任务
任何 Python 函数都可以是一个任务，我们只需要应用 @background 注释来注册它:

from background_task import background

@background(schedule=10)
def do_something(s1: str, s1: str) -> None:
   """
   Does something that takes a long time
   :param p1: first parameter
   :param p2: second parameter
   :return: None
   """
   pass

现在我们可以像往常一样在我们的项目中调用该函数:

do_something("first parameter", "second parameter")

需要注意的是，调用函数确实 不是 实际执行其代码；而是通过“django-background-tasks”模块将任务记录存储到数据库中，更准确地说是存储到“background_task”表中。出于这个原因，编写一个返回一些东西的任务函数是没有什么用的，因为无论如何该任务将在稍后的时刻在后台执行，所以在调用它时函数返回的“值”几乎没有意义.我看到的返回值的唯一用例是用于测试目的，请参阅下面的测试任务部分。
处理任务
为了实际运行已注册的任务，我们必须使用以下管理命令:

> python manage.py process_tasks

请引用 module's documentation有关命令选项的说明。
正如其他用户已经指出的那样，通常将此命令包装在 cron 作业中以确保定期处理任务。在这种情况下，duration 选项可能会很有用:它表示 process_task 命令保持运行的秒数。默认情况下，持续时间为 0，这意味着“永远运行它”，但在我看来这是非常危险的，因为如果由于某种原因命令崩溃或中断，您的任务将不再被处理，并且可能会过去很长时间你意识到了。
更好的方法是将持续时间设置为明确定义的时间，例如 15 分钟，然后配置一个 cron 作业每 15 分钟运行一次以重新启动处理命令。这样，如果命令崩溃，它无论如何都会被 cron 作业重新启动。
测试任务
通过“process_tasks”管理命令测试任务很糟糕，我们应该坚持使用 Python unittest模块，这也是“Django方式”。
当然，我不会在这篇文章中讨论单元测试，我只想指出，在单元测试期间，您希望以同步方式执行函数，就像普通的 Python 函数一样。其语法如下:

do_something.now("first parameter", "second parameter")

修饰符“now”运行函数并等待它终止。在我看来，这是返回值有用的唯一用例。有了返回值，您就可以使用 unittest 提供的“assert*”函数的全部功能。
检查任务是否已经在运行
有时可能会发生您不希望多次运行同一个任务。例如我经常使用后台任务来训练机器学习模型，这需要很多时间。为了防止我的数据被弄乱，我更愿意确保在前一个任务完成之前不能启动同一模型上的另一个训练任务。
为此，我必须在开始新任务之前检查任务是否已经在运行；但是如何唯一标识一个任务呢？对我来说，简单的方法是为任务分配一个“verbose_name”，这可以在计划任务时完成:

do_something("first parameter", "second parameter", verbose_name="my_task_verbose_name")

现在，如果我想检查这个任务是否已经在运行，我可以简单地读取 background_task 表并验证其中没有具有相同“详细名称”的任务。这可以很容易地通过利用“django-background-tasks”本身提供的任务模型来完成:

from background_task.models import Task

tasks = Task.objects.filter(verbose_name="my_task_verbose_name")
if len(tasks) == 0:
    # no task running with this name, go ahead!
    pass
else:
    # task already running
    pass

不用说，我们必须确保分配给我们任务的详细名称是唯一的。
进一步阅读
Django Background Tasks documentation