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.

Answer 1

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.