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.