如何自动设置周期性任务

Question

我的 django 应用程序之一中有一个 celery 任务。

我的

django

、

celery

和

django-celery-beat

服务在它们自己的 docker 容器中运行。由于这是一个遗留项目，我现在假设它们已经与当前配置顺利交互。

我想要的是，当应用程序部署/启动时，它会自动找到我的周期性任务并将它们注册到 django celerybeat 数据库表中。

我的任务：

# project.an_app.tasks.generate_statements
from config import celery_app

@celery_app.task(...)
def generate_statements():
    print("Statment generation runs. Current date and time:", datetime.now())
    # my task logic

这是我找到的一种解决方案。但这只有当我尝试通过管理站点访问

Periodic tasks

表时才会触发（可能会通过与这些表交互的其他方式触发，但我不想“等待”类似的事情）。

def schedule_task():
    interval, _ = IntervalSchedule.objects.get_or_create(
        every=30,
        period=IntervalSchedule.SECONDS,
    )

    PeriodicTask.objects.get_or_create(
        interval=interval,
        name="generate-statements",
        task=project.an_app.tasks.generate_statements,
    )
    
schedule_task()

查看我们的遗留代码，我发现了这个，这个看起来

from project.an_app.tasks import generate_statements

@celery_app.on_after_finalize.connect
def register_periodic_tasks(sender, **kwargs):
    """
    Callback for Celery to register periodic tasks.
    """
    sender.add_periodic_task(
        30.0,
        task=generate_statements,
        name="generate-statements",
    )

复制代码的任务确实已经是

django_celery_beat_periodictask

表了。所以我想说这有效。然而，当我重建 docker 容器时，这个版本并没有创建我的任务。

我也一直在寻找简单地将其添加到

settings.py

文件中的解决方案：

CELERY_BEAT_SCHEDULER = "django_celery_beat.schedulers:DatabaseScheduler"

CELERY_BEAT_SCHEDULE = {
    "task-name": {
        "task": "project.an_app.tasks.generate_statements",
        "schedule": timedelta(seconds=30),
    },
}

这个效果不太好。

有人可以弄清楚哪种解决方案最适合我想要的以及每种情况下我缺少什么吗？

Answer 1

查看我的 Django 应用程序，project/_init.py 中有第一个代码

from .celery import app as celery_app

__all__ = ("celery_app",)

然后我有一个芹菜.py：

import os

from celery import Celery


os.environ.setdefault("DJANGO_SETTINGS_MODULE", "core.settings")

app = Celery("core")
app.config_from_object("django.conf:settings", namespace="CELERY")
app.autodiscover_tasks()

这应该确保 celery 在项目启动时加载并且自动发现任务。

不知道这对你的情况是否有帮助。但可能值得检查。