Defining Celery tasks in Flask - Step-by-Step Execution

Concept Flow - Defining Celery tasks

Start Flask app

↓

Import Celery

↓

Create Celery instance

↓

Define task function with @app.task

↓

Call task asynchronously

↓

Worker picks task

↓

Task runs in background

↓

Result stored or returned

This flow shows how a Celery task is defined and executed asynchronously in a Flask app.

Execution Sample

Flask

from celery import Celery
app = Celery('tasks', broker='redis://localhost', backend='redis://localhost')

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

result = add.delay(4, 6)

Defines a Celery task 'add' and calls it asynchronously with arguments 4 and 6.

Execution Table

Step	Action	Evaluation	Result
1	Import Celery	Celery class imported	Celery available
2	Create Celery instance	app = Celery('tasks', broker='redis://localhost', backend='redis://localhost')	Celery app ready
3	Define task with @app.task	Function 'add' wrapped as task	Task 'add' registered
4	Call add.delay(4, 6)	Task call queued asynchronously	Async task message sent
5	Worker picks task	Worker receives task message	Task ready to run
6	Task runs: add(4, 6)	Function executes	Returns 10
7	Result stored	Result saved in backend	Result 10 available
8	Main program continues	No waiting for task	Program not blocked

💡 Task completes after worker runs function and stores result asynchronously.

Variable Tracker

Variable	Start	After Step 4	After Step 6	Final
app	None	Celery instance	Celery instance	Celery instance
add	Function	Task object	Task object	Task object
result	None	AsyncResult object (pending)	AsyncResult object (ready)	AsyncResult object (ready)

Key Moments - 3 Insights

Why doesn't calling add.delay(4, 6) run the function immediately?

Because add.delay queues the task to run asynchronously by a worker, so the main program continues without waiting. See execution_table step 4 and 8.

What does the @app.task decorator do?

It wraps the function so Celery knows it is a task to run asynchronously. See execution_table step 3.

How is the result of the task accessed?

The result is stored by Celery backend and can be accessed via the AsyncResult object returned by delay(). See execution_table step 7.

Visual Quiz - 3 Questions

Look at the execution table, what is the state of 'result' after step 4?

A None

B Function return value

C AsyncResult object (pending)

D Task function itself

💡 Hint

Check variable_tracker column 'After Step 4' for 'result'

At which step does the worker actually run the task function?

A Step 4

B Step 6

C Step 5

D Step 7

💡 Hint

See execution_table where function executes and returns value

If you remove @app.task decorator, what changes in the execution?

A add.delay() call will fail or not queue task

B Task runs synchronously immediately

C Result is stored automatically

D Worker picks task faster

💡 Hint

Look at step 3 where task registration happens

Concept Snapshot

Defining Celery tasks in Flask:
- Import Celery and create instance with broker
- Use @app.task decorator on function
- Call task asynchronously with .delay(args)
- Worker runs task in background
- Result stored for later retrieval
Key: Tasks run outside main app flow asynchronously.

Full Transcript

This visual trace shows how to define and run Celery tasks in a Flask application. First, Celery is imported and an instance is created with a broker URL. Then, a function is decorated with @app.task to mark it as a Celery task. Calling the task with .delay() queues it asynchronously without blocking the main program. A worker process picks up the task, runs the function, and stores the result. The main app can continue running while the task executes in the background. The result can be accessed later through the AsyncResult object returned by .delay(). This approach helps run time-consuming tasks outside the main request flow.

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