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

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.

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.