Periodic tasks with Celery Beat in Flask

Introduction

Periodic tasks let your app do things automatically at set times. Celery Beat helps schedule these tasks easily.

Send daily reminder emails to users.

Clean up old files every night.

Update data from an external source every hour.

Generate reports every Monday morning.

Syntax

Flask

from celery import Celery
from celery.schedules import crontab

app = Celery('tasks', broker='redis://localhost:6379/0')

app.conf.beat_schedule = {
    'task-name': {
        'task': 'module.task_function',
        'schedule': crontab(minute=0, hour=0),  # runs daily at midnight
    },
}

@app.task
def task_function():
    # task code here
    pass

Use crontab to set the schedule in a human-friendly way.

Define tasks with @app.task decorator.

Examples

This runs the send_email task every minute.

Flask

app.conf.beat_schedule = {
    'send-email-every-minute': {
        'task': 'tasks.send_email',
        'schedule': crontab(),  # runs every minute
    },
}

This runs generate_report every Monday morning at 9 AM.

Flask

app.conf.beat_schedule = {
    'weekly-report': {
        'task': 'tasks.generate_report',
        'schedule': crontab(minute=0, hour=9, day_of_week='mon'),  # every Monday at 9:00 AM
    },
}

Sample Program

This example sets up a task that prints a message every 10 seconds automatically.

Flask

from celery import Celery
from celery.schedules import crontab

app = Celery('myapp', broker='redis://localhost:6379/0')

app.conf.beat_schedule = {
    'print-hello-every-10-seconds': {
        'task': 'myapp.print_hello',
        'schedule': 10.0,  # every 10 seconds
    },
}

@app.task
def print_hello():
    print('Hello from Celery Beat!')

Important Notes

Make sure Redis or your broker is running before starting Celery and Beat.

Run celery -A myapp beat to start the scheduler.

Run celery -A myapp worker to start the worker that executes tasks.

Summary

Celery Beat schedules tasks to run automatically at set times.

Use crontab or seconds for flexible scheduling.

Tasks must be defined with @app.task and registered in 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