django-celery-beat 2.7.0

创建基于间隔的定期任务示例

要创建以间隔执行的任务，您必须首先创建间隔对象。

>>> from django_celery_beat.models import PeriodicTask, IntervalSchedule

# executes every 10 seconds.
>>> schedule, created = IntervalSchedule.objects.get_or_create(
...     every=10,
...     period=IntervalSchedule.SECONDS,
... )

这是您需要的所有字段：周期类型和频率。

您可以从一组特定的周期中进行选择

• IntervalSchedule.DAYS

• IntervalSchedule.HOURS

• IntervalSchedule.MINUTES

• IntervalSchedule.SECONDS

• IntervalSchedule.MICROSECONDS

如果您需要向用户展示，还有一个“选择元组”可用

>>> IntervalSchedule.PERIOD_CHOICES

现在我们已经定义了计划对象，我们可以创建定期任务条目

>>> PeriodicTask.objects.create(
...     interval=schedule,                  # we created this above.
...     name='Importing contacts',          # simply describes this periodic task.
...     task='proj.tasks.import_contacts',  # name of task.
... )

请注意，这是一个非常基本的示例，您还可以指定用于执行任务的参数和关键字参数，将任务发送到queue，并设置过期时间。

以下是一个指定参数的示例，注意JSON序列化是必需的

>>> import json
>>> from datetime import datetime, timedelta

>>> PeriodicTask.objects.create(
...     interval=schedule,                  # we created this above.
...     name='Importing contacts',          # simply describes this periodic task.
...     task='proj.tasks.import_contacts',  # name of task.
...     args=json.dumps(['arg1', 'arg2']),
...     kwargs=json.dumps({
...        'be_careful': True,
...     }),
...     expires=datetime.utcnow() + timedelta(seconds=30)
... )

创建基于cron的定期任务示例

crontab 调度由以下字段组成：分钟、小时、星期、月和日，因此如果您想设置一个等效于 30 * * * *（每小时后30分钟执行）的 crontab 条目，您需要指定

>>> from django_celery_beat.models import CrontabSchedule, PeriodicTask
>>> schedule, _ = CrontabSchedule.objects.get_or_create(
...     minute='30',
...     hour='*',
...     day_of_week='*',
...     day_of_month='*',
...     month_of_year='*',
...     timezone=zoneinfo.ZoneInfo('Canada/Pacific')
... )

crontab 调度通过使用 'timezone' 输入参数与特定时区相关联。

然后要使用此调度创建周期性任务，请使用与文档中早期部分基于间隔的周期性任务相同的方法，但将 interval=schedule 替换为 crontab=schedule

>>> PeriodicTask.objects.create(
...     crontab=schedule,
...     name='Importing contacts',
...     task='proj.tasks.import_contacts',
... )

暂时禁用周期性任务

您可以使用 enabled 标志暂时禁用周期性任务

>>> periodic_task.enabled = False
>>> periodic_task.save()

示例运行周期性任务

周期性任务仍然需要“工作者”来执行它们。因此请确保已安装默认的 Celery 包。（如果未安装，请按照以下说明进行安装：[https://github.com/celery/celery](https://github.com/celery/celery)）

同时运行 worker 和 beat 服务。

1. 启动 Celery worker 服务（指定您的 Django 项目名称）

   $ celery -A [project-name] worker --loglevel=info

2. 作为单独的进程，启动 beat 服务（指定 Django 调度器）

   $ celery -A [project-name] beat -l info --scheduler django_celery_beat.schedulers:DatabaseScheduler

   或者您可以使用 -S（调度器标志），更多选项请参阅 celery beat --help

   $ celery -A [project-name] beat -l info -S django

   另外，作为一个替代方案，您可以使用一个命令运行上面两个步骤（worker 和 beat 服务）（仅推荐用于开发环境）

   $ celery -A [project-name] worker --beat --scheduler django --loglevel=info

3. 现在您可以从 Django 管理界面添加和管理您的周期性任务。

从源代码下载和安装

从 http://pypi.python.org/pypi/django-celery-beat 下载 django-celery-beat 的最新版本

您可以通过以下操作进行安装

$ python3 -m venv .venv
$ source .venv/bin/activate
$ pip install --upgrade build pip
$ tar xvfz django-celery-beat-0.0.0.tar.gz
$ cd django-celery-beat-0.0.0
$ python -m build
$ pip install --upgrade .

安装后，将 django_celery_beat 添加到 Django 的设置模块

INSTALLED_APPS = [
    ...,
    'django_celery_beat',
]

使用以下命令运行 django_celery_beat 迁移

$ python manage.py migrate django_celery_beat

使用开发版本

使用 pip

您可以使用以下 pip 命令安装 django-celery-beat 的最新主要版本

$ pip install git+https://github.com/celery/django-celery-beat#egg=django-celery-beat

开发 django-celery-beat

要启动带有 Django 管理员的本地开发副本 django-celery-beat（在 http://127.0.0.1:58000/admin/），请运行

$ docker-compose up --build

以用户 admin 和密码 admin 登录。

时区感知

如果您有一个无时区感知的项目，您可以在设置文件中将 DJANGO_CELERY_BEAT_TZ_AWARE=False 设置为 False。

django-celery-beat 作为 Tidelift 订阅的一部分

django-celery-beat 的维护者以及成千上万的其他包维护者正在与 Tidelift 合作，为构建您应用程序时使用的开源依赖项提供商业支持和维护。节省时间，降低风险，提高代码质量，同时支付您使用的确切依赖项的维护者。了解更多信息：了解详情。