'Celery Beat doesn't follow schedule Fast API

This celery beat scheduler says the max tick is 5 minutes. I am developing things here so firstly it needs to automatically pick up the right times to run. Seems like it's not working correctly.

celery_beat_1    | [2022-05-12 23:18:44,224: DEBUG/MainProcess] beat: Synchronizing schedule...
celery_beat_1    | [2022-05-12 23:18:44,225: DEBUG/MainProcess] beat: Waking up in 5.00 minutes.

celery_app.py

from celery import Celery
from celery.schedules import crontab
import os


celery_app = Celery(
    "worker",
    backend=os.getenv("CELERY_RESULT_BACKEND", "redis://:3091@redis:6379/0"),
    broker=os.getenv("CELERY_BROKER_URL", "amqp://guest:guest@rabbitmq:5672//")
)

celery_app.conf.task_routes = {
    "worker.celery_worker.test_celery": "test-queue"
}

celery_app.conf.beat_schedule = {
    'celery_beat_testing': {
        'task': 'celery_app.tasks.test_beat',
        'schedule': crontab(minute='*/1')
    }
}

celery_app.conf.timezone = 'UTC'
celery_app.conf.update(task_track_started=True)

celery_worker.py

from worker.celery_app import celery_app

@celery_app.task(bind=True)
def test_celery(self):
    return ('Celery Pong: {0!r}'.format(self.request))

@celery_app.task(name='test_beat')
def test_beat():
    print('beat test', flush=True)


Sources

This article follows the attribution requirements of Stack Overflow and is licensed under CC BY-SA 3.0.

Source: Stack Overflow

Solution Source