celery の Periodic Tasks で cron 的なタスクを作成する

• 概要
• 環境
• cron タスク
• 動作確認
• 最後に
• 参考サイト

概要

celery に定期実行するためのタスクがあるので試してみました
時間は cron フォーマットで登録できるようです

環境

• macOS 11.5
• Python 3.8.3
• celery 5.1.2

cron タスク

タスクを定義し setup_periodic_tasks で add_periodic_task を使って登録できます

「実行する時刻」と「実行するタスク」を指定します オプションで name や expires が指定できます
これは apply_async で使えるオプションと同じものが指定できます

• vim sub_tasks.py

from celery import Celery
from celery.schedules import crontab

app = Celery('cron_tasks', broker='redis://localhost')

@app.on_after_configure.connect
def setup_periodic_tasks(sender, **kwargs):
    # Calls test('hello') every 10 seconds.
    sender.add_periodic_task(10.0, test.s('hello'), name='add every 10')

    # Calls test('world') every 30 seconds
    sender.add_periodic_task(30.0, test.s('world'), expires=10)

    # Executes every Monday morning at 7:30 a.m.
    sender.add_periodic_task(
        crontab(hour=7, minute=30, day_of_week=1),
        test.s('Happy Mondays!'),
    )

@app.task
def test(arg):
    print(arg)

動作確認

起動するコマンドが worker -> beat になるので注意しましょう

• pipenv run celery -A cron_tasks beat --loglevel=info

これでログを見るとちゃんとタスクが実行されているのが確認できると思います

[2021-08-12 15:39:46,142: INFO/MainProcess] Scheduler: Sending due task add every 10 (cron_tasks.test)
[2021-08-12 15:39:56,142: INFO/MainProcess] Scheduler: Sending due task add every 10 (cron_tasks.test)
[2021-08-12 15:40:06,143: INFO/MainProcess] Scheduler: Sending due task add every 10 (cron_tasks.test)
[2021-08-12 15:40:16,138: INFO/MainProcess] Scheduler: Sending due task cron_tasks.test('world') (cron_tasks.test)

最後に

cron だけでなく solar という形式でもスケジューリングできます

ワーカーが動作している状態で動的に cron タスクを追加する方法があるのか気になりました

参考サイト

• https://docs.celeryproject.org/en/stable/userguide/periodic-tasks.html