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

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

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

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

Hint

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

Practice

1. What is the main purpose of Celery Beat in a Flask application?

easy

A. To schedule tasks to run automatically at specific times

B. To handle HTTP requests asynchronously

C. To manage database migrations

D. To serve static files efficiently

Solution

1. Step 1: Understand Celery Beat's role

   Celery Beat is a scheduler that triggers tasks periodically without manual intervention.

2. Step 2: Compare with other options

   Handling HTTP requests, managing migrations, and serving static files are unrelated to scheduling tasks.

3. Final Answer:

   To schedule tasks to run automatically at specific times -> Option A

4. Quick Check:

   Celery Beat = Task scheduler [OK]

Hint: Celery Beat is all about timing tasks automatically [OK]

Common Mistakes:

• Confusing Celery Beat with request handling
• Thinking it manages database or static files
• Assuming it runs tasks only once

2. Which of the following is the correct way to define a periodic task in Celery Beat using Flask?

easy

A. @app.route('/task')\ndef my_task():\n print('Hello')

B. @celery.task\ndef my_task():\n print('Hello')

C. def my_task():\n print('Hello')

D. @celery.task\nasync def my_task():\n print('Hello')

Solution

1. Step 1: Identify task decorator

   Periodic tasks in Celery are defined with the @celery.task decorator.

2. Step 2: Check syntax correctness

   @celery.task\ndef my_task():\n print('Hello') correctly uses @celery.task and a normal function. @app.route('/task')\ndef my_task():\n print('Hello') uses Flask route decorator, which is unrelated. def my_task():\n print('Hello') lacks any decorator. @celery.task\nasync def my_task():\n print('Hello') uses async which is not standard for Celery tasks.

3. Final Answer:

   @celery.task\ndef my_task():\n print('Hello') -> Option B

4. Quick Check:

   Use @celery.task to define Celery tasks [OK]

Hint: Use @celery.task decorator for Celery tasks [OK]

Common Mistakes:

• Using Flask route decorator instead of @celery.task
• Defining tasks without any decorator
• Using async def which Celery does not support by default

3. Given this Celery Beat schedule snippet, what will happen?

beat_schedule = {
    'print-every-10-seconds': {
        'task': 'app.print_message',
        'schedule': 10,
    },
}

medium

A. The task runs once after 10 seconds

B. The task runs every 10 minutes

C. The task 'print_message' runs every 10 seconds

D. This code causes a syntax error

Solution

1. Step 1: Understand the schedule value

   The 'schedule' key with value 10 means the task runs every 10 seconds.

2. Step 2: Confirm task repetition

   Celery Beat schedules tasks repeatedly at the interval given, so it runs every 10 seconds, not once or every 10 minutes.

3. Final Answer:

   The task 'print_message' runs every 10 seconds -> Option C

4. Quick Check:

   Schedule=10 means every 10 seconds [OK]

Hint: Schedule number means seconds interval by default [OK]

Common Mistakes:

• Thinking schedule is minutes instead of seconds
• Assuming task runs only once
• Confusing syntax with crontab format

4. Identify the error in this Celery Beat configuration:

beat_schedule = {
    'task1': {
        'task': 'app.task1',
        'schedule': crontab(minute='*/5')
    },
    'task2': {
        'task': 'app.task2',
        'schedule': '10',
    }
}

medium

A. The dictionary keys 'task1' and 'task2' must be integers

B. 'task' names must be functions, not strings

C. Missing import for crontab causes runtime error

D. 'schedule' for 'task2' should be an integer or crontab, not a string

Solution

1. Step 1: Check 'schedule' types

   For 'task1', schedule uses crontab correctly. For 'task2', schedule is a string '10', which should be an integer 10 or a crontab object.

2. Step 2: Validate other parts

   Task names as strings are correct references. Import errors are not shown here. Dictionary keys can be strings.

3. Final Answer:

   'schedule' for 'task2' should be an integer or crontab, not a string -> Option D

4. Quick Check:

   Schedule must be int or crontab, not string [OK]

Hint: Schedule values must be int or crontab, not quoted strings [OK]

Common Mistakes:

• Using string instead of int for schedule interval
• Confusing task names with function objects
• Assuming dictionary keys must be integers

5. You want to run a Flask Celery task every Monday at 8:30 AM using Celery Beat. Which beat_schedule entry is correct?

hard

A. {'task': 'app.weekly_task', 'schedule': crontab(hour=8, minute=30, day_of_week='mon')}

B. {'task': 'app.weekly_task', 'schedule': crontab(hour=8, minute=30, day_of_week=2)}

C. {'task': 'app.weekly_task', 'schedule': crontab(hour=8, minute=30, day_of_week='tue')}

D. {'task': 'app.weekly_task', 'schedule': crontab(hour=8, minute=30, day_of_week='Monday')}

Solution

1. Step 1: Understand crontab day_of_week format

   Celery's crontab accepts day_of_week as string names like 'mon', 'tue', or integers 0-6 (0=Sunday).

2. Step 2: Evaluate options

   {'task': 'app.weekly_task', 'schedule': crontab(hour=8, minute=30, day_of_week='mon')} uses 'mon' string which is correct. {'task': 'app.weekly_task', 'schedule': crontab(hour=8, minute=30, day_of_week=2)} uses integer 2 which is Tuesday (0=Sunday). {'task': 'app.weekly_task', 'schedule': crontab(hour=8, minute=30, day_of_week='tue')} uses 'tue' string which is Tuesday. {'task': 'app.weekly_task', 'schedule': crontab(hour=8, minute=30, day_of_week='Monday')} uses 'Monday' full name which is not supported.

3. Final Answer:

   {'task': 'app.weekly_task', 'schedule': crontab(hour=8, minute=30, day_of_week='mon')} -> Option A

4. Quick Check:

   Use 'mon' string for Monday in crontab [OK]

Hint: Use three-letter lowercase day names like 'mon' for crontab [OK]

Common Mistakes:

• Using full day names like 'Monday' instead of 'mon'
• Using string numbers like '1' instead of integers or names
• Confusing day_of_week numbering starting at Sunday