Defining tasks in Django - Practice Problems & Coding Challenges

Challenge - 5 Problems

❓ component_behavior

intermediate

What is the output of this Django Celery task when called synchronously?

Consider this Django Celery task definition and its synchronous call. What will be printed?

Django

from celery import shared_task

@shared_task
def add(x, y):
    return x + y

result = add(4, 5)
print(result)

A Raises TypeError because task is not awaited

B Prints <AsyncResult: ...>

C Prints 9

D Prints None

💡 Hint

Calling the task function directly runs it like a normal function.

When you call a Celery task function directly (not using .delay() or .apply_async()), it runs immediately and returns the result. So print(result) outputs 9.

❓ state_output

intermediate

What is the type of result when calling a Celery task asynchronously?

Given this task call, what is the type of the variable 'result'?

Django

from celery import shared_task

@shared_task
def multiply(x, y):
    return x * y

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

A <class 'celery.result.AsyncResult'>

B <class 'int'>

C <class 'NoneType'>

D <class 'str'>

💡 Hint

Using .delay() returns a special object representing the task execution.

Calling .delay() on a Celery task returns an AsyncResult object that tracks the task status and result asynchronously.

📝 Syntax

advanced

Which option correctly defines a periodic task in Django Celery?

Select the correct way to define a periodic task that runs every 10 minutes using Celery beat.

A

@shared_task
def cleanup():
    pass

app.conf.beat_schedule = {
    'cleanup-every-10-minutes': {
        'task': 'app.tasks.cleanup',
        'schedule': timedelta(minutes=10),
    },
}

B

@shared_task
@periodic_task(run_every=10)
def cleanup():
    pass

C

@shared_task
def cleanup():
    pass

app.conf.beat_schedule = {
    'cleanup-every-10-minutes': {
        'task': 'app.tasks.cleanup',
        'schedule': 600,
    },
}

D

@shared_task
@periodic_task(run_every=timedelta(minutes=10))
def cleanup():
    pass

💡 Hint

The schedule value must be a timedelta object, not an integer.

The correct way is to define the task normally and configure the beat_schedule with a timedelta for the schedule. Option A uses timedelta(minutes=10) correctly.

🔧 Debug

advanced

Why does this Celery task raise an ImportError when called?

Given this task code, why does calling task.delay() raise ImportError: No module named 'myapp.tasks'?

Django

from celery import shared_task

@shared_task
def send_email():
    from myapp.utils import send
    send()

# Called from another module
send_email.delay()

A Celery tasks cannot import inside the function body

B The import inside the task function causes circular import

C The task decorator is missing parentheses

D The task module is not in the Python path or installed apps

💡 Hint

Check if the module 'myapp.tasks' is discoverable by Celery worker.

Celery workers must be able to import the task module. If 'myapp.tasks' is not in the Python path or not listed in INSTALLED_APPS, ImportError occurs.

🧠 Conceptual

expert

What happens if a Celery task raises an exception during execution?

When a Celery task raises an exception, which of the following is true about the task state and result?

A The task silently retries without changing state

B The task state is set to 'FAILURE' and the exception info is stored in the result

C The task state remains 'PENDING' and no error info is stored

D The task state is set to 'SUCCESS' but result is None

💡 Hint

Celery tracks task failures explicitly.

When a task raises an exception, Celery marks it as 'FAILURE' and stores the exception details for inspection.

Practice

1. What decorator is commonly used to define a background task in Django with Celery?

easy

A. @task_runner

B. @shared_task

C. @async_task

D. @background_task

Solution

1. Step 1: Understand task definition in Django with Celery

   Celery uses the @shared_task decorator to mark functions as tasks that can run asynchronously.

2. Step 2: Identify the correct decorator

   Among the options, only @shared_task is the correct and standard decorator for defining tasks.

3. Final Answer:

   @shared_task -> Option B

4. Quick Check:

   Task decorator = @shared_task [OK]

Hint: Remember: Celery tasks use @shared_task decorator [OK]

Common Mistakes:

• Using @background_task which is not a Celery decorator
• Confusing @async_task with async/await syntax
• Using @task_runner which is not valid in Django

2. Which of the following is the correct way to call a Celery task asynchronously in Django?

easy

A. my_task.run()

B. my_task.execute()

C. my_task.delay()

D. my_task.start()

Solution

1. Step 1: Recall how to call Celery tasks asynchronously

   Celery tasks are called asynchronously using the delay() method on the task function.

2. Step 2: Identify the correct method

   Only delay() triggers the task asynchronously; other methods like run() execute synchronously or do not exist.

3. Final Answer:

   my_task.delay() -> Option C

4. Quick Check:

   Async call method = delay() [OK]

Hint: Use delay() to run tasks asynchronously [OK]

Common Mistakes:

• Calling run() which runs task synchronously
• Using execute() which is not a Celery method
• Trying start() which does not exist for tasks

3. Given this task definition:

from celery import shared_task

@shared_task
def add(x, y):
    return x + y

result = add.delay(4, 5)

What will result.get() return?

medium

A. 9

B. None

C. An AsyncResult object

D. A syntax error

Solution

1. Step 1: Understand the task and its call

   The add function adds two numbers. Calling add.delay(4, 5) runs it asynchronously and returns an AsyncResult.

2. Step 2: Using result.get() retrieves the task result

   Calling result.get() waits for the task to finish and returns the sum, which is 9.

3. Final Answer:

   9 -> Option A

4. Quick Check:

   Task result = 9 [OK]

Hint: delay() returns AsyncResult; get() fetches the actual result [OK]

Common Mistakes:

• Thinking delay() returns the result immediately
• Confusing AsyncResult object with the actual result
• Expecting None because task runs asynchronously

4. Identify the error in this task definition:

from celery import shared_task

@shared_task
def multiply(x, y):
return x * y

medium

A. Function name is invalid

B. Missing @shared_task decorator

C. Using delay() incorrectly

D. Indentation error in function body

Solution

1. Step 1: Check the function syntax

   The function body must be indented inside the function definition. Here, return x * y is not indented.

2. Step 2: Identify the error type

   Python requires indentation for blocks. Missing indentation causes an IndentationError.

3. Final Answer:

   Indentation error in function body -> Option D

4. Quick Check:

   Python blocks need indentation [OK]

Hint: Check indentation inside function definitions [OK]

Common Mistakes:

• Ignoring indentation errors
• Assuming decorator is missing when it is present
• Confusing function name validity with syntax errors

5. You want to define a task that sends emails but only if the email address is not empty. Which of these task definitions correctly applies this condition?

hard

A. from celery import shared_task

@shared_task
def send_email(email):
if email:
# send email code
return 'Sent'
return 'No email provided'

B. from celery import shared_task

@shared_task
def send_email(email):
if email == None:
return 'No email provided'
# send email code
return 'Sent'

C. from celery import shared_task

@shared_task
def send_email(email):
if not email:
return 'Sent'
# send email code
return 'No email provided'

D. from celery import shared_task

def send_email(email):
if email:
# send email code
return 'Sent'
return 'No email provided'

Solution

1. Step 1: Check for correct task decorator and condition

   from celery import shared_task @shared_task def send_email(email): if email: # send email code return 'Sent' return 'No email provided' uses @shared_task and checks if email: which correctly tests for a non-empty email.

2. Step 2: Verify logic correctness

   from celery import shared_task @shared_task def send_email(email): if email: # send email code return 'Sent' return 'No email provided' returns 'Sent' only if email is truthy (not empty), else returns 'No email provided'. This matches the requirement.

3. Final Answer:

   Option A correctly defines the task with the condition -> Option A

4. Quick Check:

   Use if email: to check non-empty string [OK]

Hint: Use if email: to check non-empty strings in tasks [OK]

Common Mistakes:

• Forgetting @shared_task decorator
• Using if not email: incorrectly reversing logic
• Not indenting task function properly