Setting up a CRON (or Scheduled) Job Programatically

Introduction

Nearly every application that I have developed in the past has a requirement for Scheduled Jobs. In the Unix world, they call it CRON jobs. Even if you are using Windows, you will most likely still be able to run CRON jobs using some language-specific implementation or library. I use Celery for Python and Windows.

Having developed systems using various platforms/languages before, one thing I can say is that the implementations are all similar, meaning that it becomes very easy to implement a scheduled job in any other platform once you know the logic to do so.

With the foregoing statement, I am just saying the obvious anyway because a programming language or platform is just a tool based on fundamental principles. Learn to do something in one programming language, you will certainly do it very easily in another. Fundamentals such as variables, data structures, control constructs (loops, conditions), objects, functions, etc all follow same principles. Any programmer who has experience in many programming languages will tell you the same.

Today, I want to share with you how you can set up a scheduled job (CRON) in Python using Celery. The example I share will aim to do a simple but important task as I explain in the next paragraph of the requirement statement.


The Requirement

There is an existing web application used for the management of customer data for a company. A new requirement has emerged. To make a promotional offer, the client requires that the system should send automated emails with a list of customers whose birthdates (anniversary) are within the next one month. Such an email should be sent automatically to Management every first day of the month so that Management can make a promotional offer.

The above requirement is typical of the real world, and it must be solved programmatically by ultimately translating it into code… and it’s pretty easy to do so. Always remember that a good programmer is one who can translate real-world problems or requirements into code that solves the problem at hand.

Without wasting more time, let me now show you how you can solve this problem in Python, Flask and Celery. If you understand the principles, then you can implement it in any other language/platform.

We shall make some assumptions:

1. A database with data on customers and transactions already exists; also customer birthdates exist in the database.

2. Flask and SQLAlchemy have been setup to implement the above database.

3. Celery and any associated requirements have been installed.

4. sendEmail function has been set up.


The Solution

Having made the above assumptions, we can now embark on the following. The important function is called getAnniversaryList. Please pay more attention to the bolded sections of the code below. Apologies that the code is rather verbose since I picked it directly from my codebase.

Step 1: set up the schedule

celery.conf.beat_schedule = {
    'create_task': { #some other task doing something else - just for illustration
        'task': 'create_task',
        #'schedule': 60.0, #every 60 seconds
        #'schedule': crontab(hour=7, minute=30, day_of_week=1), # Executes every Monday morning at 7:30 a.m.
        'schedule': crontab(minute='*',hour='*', day_of_week='sun'), #every minute on saturday
        'args': ('1')
    },
    'getAnniversaryList': { #our task to solve our problem at hand
        'task': 'getAnniversaryList',
        'schedule': crontab(hour=9, minute=15, day_of_month='1'), # Executes every first day of month at 9.15
        'args' : (True,), #True means scheduled, so send email
    },
}

Step 2:write the code for the job (task)

#This task (function) is decorated with celery.task to make it a background job. Decorator adds some functionality to an object/function.
@celery.task(name="getAnniversaryList")
@disable_job #user-defined decorator
def getAnniversaryList(scheduled): #for scheduled task, scheduled is True
    from etoolbox.models import db, Customer, CustomerProfile
    from . import get_emails
    developer_email, sender_email = get_emails()

    #monitoring on tables
    results = '<span style="font-size: 20px; text-align: center; width: 100%;">Anniversary List</span><br/><br/>'
    results += "<table border='1' style='border-collapse: collapse; width: 100%;'>"
    results += "<caption>Anniversary List</caption>"
    results += "<thead class='highlight'>"
    results += "<tr> \
            <th>#</th> \
            <th>Customer Name</th> \
            <th>Join Date</th> \
            <th>Weekday</th> \
            <th>ISO Weekday</th> \
            <th>Days Remaining</th> \
        </tr>"

    results += "</thead>"

    today = datetime.today()
    days_ago = today - timedelta(days=1)
    #age = (datetime.now()-rows.last_updated).days
    one_month_from_now = today + timedelta(days=30)

    query = db.session.query(Customer) \
        .filter(and_(Customer.birth_date >= today, Customer. birth_date < one_month_from_now)) \
        .with_entities(Customer.record_name, Customer. birth_date) \
        .order_by(Customer. birth_date.desc())

    for index, row in enumerate(query, start=1):
        results += f"<tr> \
                <td>" + f"{index:2d}" + "</td> \
                <td>" + f"{row[0]}" + "</td> \
                <td>" + row[1].strftime('%Y-%m-%d') + "</td> \
                <td>" + f"{str(row[1].weekday())} - {row[1].strftime('%a')} - {row[1].strftime('%A')}" + "</td> \
                <td>" + f"{str(row[1].isoweekday())} - {row[1].strftime('%a')} - {row[1].strftime('%A')}" + "</td> \
                <td>" + f"{(row[1] - date.today()).days}" + "</td> \
            </tr>"

    results += "</table>"

    reportOrientation = 'Portrait'
    headerCenter = 'Anniversary'
    footerCenter = f"Generated by {('system' if scheduled else session.get('user_email'))} on {datetime.now()}"
    fileName = 'Anniversary-' + ('system' if scheduled else session.get('user_email')) + '.pdf'

    toPdf = saveToPdf.apply_async(args=[results, reportOrientation, headerCenter, footerCenter, fileName], countdown=0)

    if scheduled: #scheduled
        #now email the file
        to = developer_email
        toSendEmail = sendEmail.apply_async(args=[to, results, None], kwargs={'subject': 'Anniversary List'}, countdown=1)
        return {"status": True}  # Change is here

    return results

With the above setup, the task will be executed every first day of the month at 9.15. It will send an email with a formatted HTML list of all customers with birthdates falling within the next one month.

Probably the most important sections of the code snippets are the bolded ones.