如何使用 django-background-tasks答案

【问题标题】：how to use django-background-tasks如何使用 django-background-tasks
【发布时间】：2015-08-29 05:38:11
【问题描述】：

我正在制作一个 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 django background-task

【解决方案1】：

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

环境

Python 3.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”表中。因为这个原因，写一个返回一些东西的任务函数是没有多大用处的，因为无论如何任务都会在稍后的时刻在后台执行，所以函数在被调用时返回的“值”几乎是没有意义的.我看到的返回值的唯一用例是用于测试目的，请参阅下面的测试任务部分。

处理任务

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

> python manage.py process_tasks

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

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

测试任务

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

我当然不会在这篇文章中讨论 unittest，我只想指出，在单元测试期间，您希望以同步方式执行函数，就像正常的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”本身提供的 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

【讨论】：

谢谢。官方文档中缺少“检查任务是否已在运行”
@BenediktS.Vogler 我知道，但我需要这个功能，因为我必须浏览包代码才能弄清楚，所以我想分享这个。很高兴它有帮助。
在您关于“检查任务是否已在运行”@SalBorrelli 的示例中，如果该过滤器是从计划任务函数中运行的，那么该函数是否可以访问其自己的详细名称 - 外部必须运行一个单独的过滤器才能找到自己？
@txbarnesx 当然，我不知道您的特定用例的详细信息，但是如果您需要从函数中访问计划任务的详细名称，那么没有什么可以阻止您将其作为我猜是附加参数，比如def do_something(s1: str, s1: str, name: str) -> None:，然后是do_something("first parameter", "second parameter", "my_task_verbose_name", verbose_name="my_task_verbose_name")。我很确定一定有更好的方法，也许是通过 kwargs 或某种方式，但上面的解决方案应该可以。如果您找到更好的方法，请告诉我。
@TanvirAhmed 如果您的生产服务器是 Linux 机器，我相信最简单的方法是为其配置一个 cron 作业。例如，将此行添加到 crontab 以每 30 分钟运行一次process_tasks：*/30 * * * * . /path/to/your/.service

【解决方案2】：

django-background-task 和 django-background-tasks 是有区别的。 django-background-task 未维护并且与较新的 Django 版本不兼容。不久前，我们使用新功能对其进行了更新和扩展，并在Github 上维护了新的向后兼容包 django-background-tasks。可以从PyPI 下载或安装新的 django-background-tasks 应用。

【讨论】：

嘿，我正在为我的项目使用 Django-background-tasks。存在多次重复任务的问题。你知道为什么会这样吗？我正在用 4 个工作人员运行我的 django 服务器。这会影响它吗？

【解决方案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 分钟或您想要的任何间隔调用一次。

【讨论】：

【解决方案4】：

您应该按照here 的描述运行python manage.py process_tasks。您可以将其添加到 crontab 以定期执行。

更新：

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

【讨论】：

好的，这意味着我无法在模型或视图本身中安排任务
@Rohit 我认为您可以将此方法保留在 models.py 中（单独的文件只是文档中的建议），但作为带有 id 参数的全局函数，因为任务参数以 json 格式存储在数据库中所以你不能传递你的对象，因为它以后不能从 json 复制。
当我运行 python manage.py process_tasks，我得到这个错误： C:\Python34\lib\site-packages\background_task\models.py:28: RemovedInDjango18War ning: Manager.get_query_set 方法应该是重命名get_queryset。类TaskManager（models.Manager）：未知命令：'process_tasks'我在这里缺少什么
@Rohit 要公开其命令，您应该将 'background_task' 添加到 INSTALLED_APPS。
是的.. 完成了.. 运行迁移，python manage.py process_tasks 工作正常.. 在 tasks.py 我有这个：@background(schedule=5) def create_node(): user = User .objects.get(id=4) Node.objects.create(post='it's a time based node', user=user, category='F') 如何将任务添加到数据库中的后台任务表？