How to set a (Django) celery beat cron schedule from string value

Question

I am using Django with celery beat. I would like to configure the cron schedule via env var (string value of the cron).

We are currently setting the cron schedule like this, using celery.schedules.crontab:

CELERY_BEAT_SCHEDULE = {
    "task_name": {
        "task": "app.tasks.task_function",
        "schedule": crontab(minute=0, hour='1,3,5,13,15,17,19,21,23'),
    },
}

I would like to configure the schedule by passing a crontab in string format, like this:

CELERY_BEAT_SCHEDULE = {
    "task_name": {
        "task": "app.tasks.task_function",
        "schedule": '0 15 10 ? * *',
    },
}

However, I cannot find any documentation for how I can do this. The naive attempt above does not work. I could parse the cron string into minute/hour/day etc values, and then pass them into crontab with the relevant kwargs, but that feels quite messy. It seems there should be an elegant way to do this.

Mohammad Golam Dostogir · Accepted Answer · 2024-01-10 00:57:23Z

From my knowledge I think there are no straight way to do this apart from creating a custom function which meets this requirements.

Here is how you can do this.

Create a new file if you do not already have like utils.py/helpers.py or whatever suits for your project and add this custom method into it.

def parse_cron_string(cron_str):
    parts = cron_str.split()
    
    minute, hour, day_of_month, month, day_of_week = parts[:5] #first five elements from the split cron string
    
    day_of_month = '*' if day_of_month == '?' else day_of_month
    
    month = '*' if month == '?' else month
    
    day_of_week = '*' if day_of_week == '?' else day_of_week
    
    return {
        'minute': minute,
        'hour': hour,
        'day_of_month': day_of_month,
        'month_of_year': month,
        'day_of_week': day_of_week,
    }

Now call this newly created method in your suitable settings file:

import os
from celery.schedules import crontab
from .utils import parse_cron_string  # Adjust the import path as needed

CRON_STRING = os.getenv('CRON_SCHEDULE', '0 15 10 ? * *')
schedule_values = parse_cron_string(CRON_STRING)

CELERY_BEAT_SCHEDULE = {
    "task_name": {
        "task": "app.tasks.task_function",
        "schedule": crontab(**schedule_values),
    },
}

Please note this is an example of how you can achieve this you may need to adjust as per your needs but as you describe this should meet the requirements.

Collectives™ on Stack Overflow

How to set a (Django) celery beat cron schedule from string value

1 Answer 1

Comments

Your Answer

Hot Network Questions

Collectives™ on Stack Overflow

1 Answer 1

Comments

Your Answer

Sign up or log in

Post as a guest

Related