如何使用 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”的快速备忘单的正确位置。希望我不会是唯一一个使用它的人:)

环境

• 蟒蛇 3.8
• 姜戈 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

有关命令选项的说明，请参阅模块的文档。正如其他用户已经指出的那样，通常将此命令包装在 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 后台任务文档

Answer 2

有Django的后台任务和Django的后台任务之间的差异小号。django-background-task 未维护且与较新的 Django 版本不兼容。不久前，我们使用新功能对其进行了更新和扩展，并在Github上维护了新的向后兼容包 django-background-task s。可以从PyPI下载或安装新的 django-background-tasks 应用程序。

Answer 3

您似乎使用了错误的方式。

假设您必须执行一些特定的任务，例如在用户注册后 5 分钟发送邮件。那么你要做的是：

使用 django-background-task 创建任务。

@background(schedule=60*5)
def send_html_mail_post(id, template):
    u = User.objects.get(id=id)
    user_email = u.email
    subject = "anything"
    html_content = template.format(arguments)
    from_email, to = from_email, user_email
    text_content = ''
    msg = EmailMultiAlternatives(subject, text_content, from_email, [to])
    msg.attach_alternative(html_content, "text/html")
    msg.send()

顶部的装饰器定义了实际事件将在函数被调用多少时间后发生。

当你需要的时候调用它。

def create_user_profile(sender, instance, created, **kwargs):
    if created:
        up = UserProfile.objects.create(user=instance)
        up.save()
        tasks.send_welcome_email(up.user.id, template=template)

这将创建任务并将其保存在数据库中，并将执行时间存储在数据库中。

您想做的事情，定期做某事，可以通过创建 cron 作业更轻松地完成。

您所做的是创建一个如问题中所示的函数。然后定义一个 cron 作业，每 5 分钟或您想要的任何时间间隔调用它。