如何使用 django-background-tasks

how to use django-background-tasks

我正在制作一个 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 来安排任务或者让我参考一些现有的文档。

您应该 运行 python manage.py process_tasks 描述 here。可以加入crontab定时执行

更新：

  1. 你不需要 运行 process_tasks 使用 crontab 因为这个命令在内部每 5 秒休眠一次（这个值是可配置的）然后再次检查是否有任何任务要 运行.
  2. 你的任务看起来很奇怪。您应该在单独的文件中将其声明为全局函数，并在其中传递 id 模型，通过 id 获取对象进行计算并保存您的对象。

你好像用错了。

假设您必须执行某些特定任务，比如在用户注册后 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 分钟或您想要的任何时间间隔调用一次。

django-background-task 和 django-background-task 之间有区别s。 django-background-task 无人维护且与较新的 Django 版本不兼容。我们不久前用新功能更新和扩展了它，并在 Github. The new django-background-tasks app can be downloaded or installed from the PyPI.

上维护了新的向后兼容包 django-background-tasks

由于这个问题似乎很笼统，根据我的个人经验，我认为这是快速作弊 sheet 关于“如何使用 django-background-tasks” 的正确位置。希望我不会是唯一一个使用它的人:)

环境

  • Python3.8
  • Django 3.1

安装

我喜欢 pipenv 所以:

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

现在在settings.py中将'background_task'添加到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”table。出于这个原因，编写一个 returns 东西的任务函数没什么用，因为无论如何该任务稍后都会在后台执行，所以“值” return 由调用时的函数几乎没有意义。我看到的 return 值的唯一用例是用于测试目的，请参阅下面的 测试任务 部分。

处理任务

为了实际 运行 注册任务，我们必须使用以下管理命令：

> python manage.py process_tasks

有关命令选项的说明，请参阅module's documentation。 正如其他用户已经指出的那样，通常将此命令包装在 cron 作业中以确保定期处理任务。在这种情况下，duration 选项可能会很有用：它表示 process_task 命令保持 运行ning 的秒数。默认情况下，持续时间为 0，这意味着“运行 它永远”，但在我看来这是相当冒险的，因为如果由于某种原因命令崩溃或被中断，您的任务将不会再被处理并且很长一段时间时间可能在你意识到之前就过去了。

更好的方法是将持续时间设置为明确定义的时间，例如 15 分钟，然后将 cron 作业配置为每 15 分钟 运行 以重新启动处理命令。这样，如果命令崩溃，它会在稍后由 cron 作业重新启动。

测试任务

通过“process_tasks”管理命令测试任务很糟糕，我们应该坚持使用 Python unittest 模块，这也是“Django 方式”。

我不打算在这个 post 讨论 unittest 当然，我只想指出在单元测试期间你想执行的功能以同步方式，就像普通的 Python 函数一样。其语法如下：

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

修饰符“now”运行函数并等待它终止。在我看来，这是 return 值有用的唯一用例。使用手头的 return 值，您可以使用 unittest.

提供的“assert*”函数的全部功能

正在检查任务是否已经 运行ning

有时您可能不希望同一任务被 运行 多次。例如，我经常使用后台任务来训练机器学习模型，这会花费很多时间。为了防止我的数据被弄乱，我更愿意确保在同一模型上的另一项训练任务不能在前一个任务完成之前开始。

为此，我必须在开始新任务之前检查任务是否已经 运行ning；但是如何唯一标识一个任务呢？对我来说，简单的方法是给任务分配一个“verbose_name”，这可以在任务安排的时候完成：

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

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

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