Periodic tasks with Celery Beat in Flask - Mini Project: Build & Apply

Periodic tasks with Celery Beat

📖 Scenario: You are building a Flask web app that needs to perform a task repeatedly at fixed time intervals, like sending reminder emails every minute.To do this, you will use Celery with Celery Beat to schedule periodic tasks.

🎯 Goal: Build a Flask app with Celery and Celery Beat that runs a simple periodic task every minute.

📋 What You'll Learn

Create a Celery app instance linked to Flask

Configure Celery Beat to schedule a periodic task

Define a simple task function that prints a message

Set the periodic task to run every minute

💡 Why This Matters

🌍 Real World

Periodic tasks are common in web apps for sending notifications, cleaning databases, or syncing data automatically.

💼 Career

Understanding Celery and Celery Beat is valuable for backend developers working with Python web frameworks to implement background and scheduled jobs.

1

Set up Flask and Celery app

Create a Flask app instance called app and a Celery app instance called celery with the broker URL set to redis://localhost:6379/0.

Flask

# Your code here

Need a hint?

Import Flask and Celery. Create app = Flask('app'). Then create celery = Celery(app.name, broker='redis://localhost:6379/0').

2

Configure Celery Beat schedule

Add a configuration dictionary called celery.conf.beat_schedule to schedule a periodic task named 'print-every-minute' that runs every 60 seconds.

Flask

from flask import Flask
from celery import Celery

app = Flask('app')
celery = Celery(app.name, broker='redis://localhost:6379/0')

# Your code here

Need a hint?

Use celery.conf.beat_schedule to define a dictionary with the key 'print-every-minute'. Set 'schedule' to 60.0 seconds.

3

Define the periodic task function

Define a Celery task function called print_message using the @celery.task decorator that prints 'Periodic task executed'.

Flask

from flask import Flask
from celery import Celery
from celery.schedules import crontab

app = Flask('app')
celery = Celery(app.name, broker='redis://localhost:6379/0')

celery.conf.beat_schedule = {
    'print-every-minute': {
        'task': 'app.print_message',
        'schedule': 60.0
    }
}

# Your code here

Need a hint?

Use @celery.task decorator. Define def print_message(): and inside it call print('Periodic task executed').

4

Complete Celery app with Flask context

Add the line celery.conf.timezone = 'UTC' to set the timezone for Celery Beat scheduling.

Flask

from flask import Flask
from celery import Celery
from celery.schedules import crontab

app = Flask('app')
celery = Celery(app.name, broker='redis://localhost:6379/0')

celery.conf.beat_schedule = {
    'print-every-minute': {
        'task': 'app.print_message',
        'schedule': 60.0
    }
}

@celery.task
def print_message():
    print('Periodic task executed')

# Your code here

Need a hint?

Set the timezone by adding celery.conf.timezone = 'UTC' after the beat schedule.