set

from celery.schedules import crontab
from mysite import celery_app  


@celery_app.on_after_finalize.connect
def (sender, **kwargs):

    sender.add_periodic_task(
        schedule=crontab(minute="*/1"),  # 任务日程，每分钟执行一次
        sig=debug_periodic_task.s("[Debug periodic task]"),  # 信号处理函数
        args=("[Debug periodic task]",)  # 必须是一个元组，如果只有一个元素在元组中，得在这个元素后面加上逗号。
        name="debug periodic task",
    )


@celery_app.task
def debug_periodic_task(args):
    print(args)

from celery import shared_task


@shared_task
def debug_periodic_task(args):
    print(args)

from celery.schedules import crontab

CELERY_BEAT_SCHEDULE = {
    "debug-periodic-task": {
        "task": "app_name.tasks.debug_periodic_task",
        "schedule": crontab(minute="*/1"),
        "args": ("[Periodic task]",),
    },
}