Defining Celery tasks in Flask - Cheat Sheet & Quick Revision

Recall & Review

beginner

What is a Celery task in Flask?

A Celery task is a function that runs asynchronously in the background, outside the main Flask app, to handle time-consuming jobs without blocking user requests.

beginner

How do you define a simple Celery task in Flask?

You define a Celery task by decorating a Python function with @celery.task or @celery.task(), which tells Celery to run it asynchronously.

intermediate

Why should Celery tasks be idempotent?

Celery tasks should be idempotent because they might run more than once if retries happen. Idempotent means running the task multiple times has the same effect as running it once.

intermediate

What is the role of the Celery app instance when defining tasks?

The Celery app instance connects your Flask app to the Celery worker. You use it to register tasks and configure how tasks run in the background.

beginner

How do you call a Celery task asynchronously after defining it?

You call a Celery task asynchronously by using the .delay() method on the task function, like my_task.delay(args), which queues the task for background execution.

Which decorator is used to define a Celery task in Flask?

A @celery.task

B @flask.task

C @app.route

D @async.task

The correct decorator to define a Celery task is @celery.task, which marks a function to run asynchronously.

What method do you use to run a Celery task asynchronously?

A run()

B delay()

C execute()

D start()

You use the delay() method to queue a Celery task for asynchronous execution.

Why is it important for Celery tasks to be idempotent?

A To improve UI responsiveness

B To reduce memory usage

C Because tasks may run multiple times due to retries

D To speed up task execution

Celery tasks might be retried and run multiple times, so idempotency ensures repeated runs don't cause errors or unwanted effects.

What does the Celery app instance do in a Flask project?

A Manages background task execution

B Stores user sessions

C Renders HTML templates

D Handles HTTP requests

The Celery app instance manages how tasks run in the background, separate from the Flask web server.

Which of these is NOT a good practice when defining Celery tasks?

A Using the @celery.task decorator

B Making tasks idempotent

C Keeping tasks short and focused

D Running long blocking code inside tasks

Long blocking code should be avoided inside tasks to keep workers responsive; tasks should be short and efficient.

Explain how to define and call a Celery task in a Flask application.

Think about how you tell Celery which function to run in the background and how you start it.

Why should Celery tasks be idempotent and how does this affect task design?

Consider what happens if a task fails and Celery tries again.

Practice

1. What is the main purpose of defining a Celery task in a Flask application using @celery_app.task?

easy

A. To make the Flask app run faster by disabling routes

B. To automatically restart the Flask server

C. To run functions asynchronously in the background

D. To create database models

Solution

1. Step 1: Understand Celery task decorator

   The @celery_app.task decorator marks a function to run as a background task outside the main Flask process.

2. Step 2: Purpose of asynchronous tasks

   This allows long-running or slow tasks to run without blocking the main app, improving responsiveness.

3. Final Answer:

   To run functions asynchronously in the background -> Option C

4. Quick Check:

   Celery tasks = background async functions [OK]

Hint: Celery tasks run outside Flask to avoid blocking [OK]

Common Mistakes:

• Thinking Celery tasks restart the server
• Confusing tasks with database models
• Believing tasks disable routes

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

easy

A. @celery_app.task
def my_task(): pass

B. celery_app.run_task(my_task)

C. @app.route('/task')
def my_task(): pass

D. def my_task(): pass

Solution

1. Step 1: Identify the decorator for Celery tasks

   The correct decorator to define a Celery task is @celery_app.task.

2. Step 2: Check syntax correctness

   @celery_app.task def my_task(): pass correctly uses @celery_app.task above the function definition.

3. Final Answer:

   @celery_app.task\ndef my_task(): pass -> Option A

4. Quick Check:

   Use @celery_app.task to define tasks [OK]

Hint: Look for @celery_app.task decorator above function [OK]

Common Mistakes:

• Using Flask route decorator instead of Celery task
• Calling run_task instead of using decorator
• Defining function without decorator

3. Given the following code, what will be the output when add.delay(4, 5) is called?

from celery import Celery
celery_app = Celery('tasks')

@celery_app.task
def add(x, y):
    return x + y

result = add.delay(4, 5)
print(type(result))

medium

A. <class 'celery.result.AsyncResult'>

B. <class 'int'>

C. 9

D. None

Solution

1. Step 1: Understand what delay() returns

   The delay() method queues the task and returns an AsyncResult object, not the actual result.

2. Step 2: Check the printed type

   Printing type(result) shows the class of the returned object, which is celery.result.AsyncResult.

3. Final Answer:

   <class 'celery.result.AsyncResult'> -> Option A

4. Quick Check:

   delay() returns AsyncResult object [OK]

Hint: delay() returns AsyncResult, not the task output [OK]

Common Mistakes:

• Expecting delay() to return the task result immediately
• Confusing AsyncResult with the actual return value
• Printing result instead of type(result)

4. Identify the error in the following Celery task definition:

from celery import Celery
celery_app = Celery('tasks')

@celery_app.task()
def multiply(x, y):
    return x * y

result = multiply.delay(3, 7)
print(result.get())

medium

A. Using parentheses in @celery_app.task() decorator causes error

B. Calling delay() instead of apply_async() is wrong

C. Missing Flask app context inside the task

D. No error; code runs correctly

Solution

1. Step 1: Check decorator syntax

   Using parentheses in @celery_app.task() is allowed and works correctly.

2. Step 2: Verify delay() usage and result retrieval

   Calling delay() queues the task, and result.get() waits for and returns the result properly.

3. Final Answer:

   No error; code runs correctly -> Option D

4. Quick Check:

   Parentheses in decorator are optional and valid [OK]

Hint: Parentheses in @celery_app.task() are optional [OK]

Common Mistakes:

• Thinking parentheses cause syntax error
• Confusing delay() with apply_async()
• Assuming Flask context is always needed in tasks

5. You want to define a Celery task in Flask that sends emails but only if the recipient email is not empty. Which of the following task definitions correctly applies this condition inside the task?

hard

A. @celery_app.task
def send_email(email):
if email is not None:
print(f"Sending email to {email}")

B. @celery_app.task
def send_email(email):
if not email:
return
print(f"Sending email to {email}")

C. @celery_app.task
def send_email(email):
if email is None:
print("No email provided")
else:
print(f"Sending email to {email}")

D. @celery_app.task
def send_email(email):
print(f"Sending email to {email}")

Solution

1. Step 1: Understand the condition to skip empty emails

   The task should not send emails if the email is empty or falsy (like empty string).

2. Step 2: Check which option correctly skips empty emails

   @celery_app.task def send_email(email): if not email: return print(f"Sending email to {email}") uses if not email: return to exit early, preventing sending to empty emails.

3. Final Answer:

   if not email: return to skip empty emails -> Option B

4. Quick Check:

   Use if not email: return to skip empty input [OK]

Hint: Return early if email is empty to skip sending [OK]

Common Mistakes:

• Not checking for empty string, only None
• Sending email without any condition
• Using incorrect condition that doesn't skip empty