Defining Celery Tasks — Complete Guide
In this tutorial, you will learn about Defining Celery Tasks. We cover key concepts, practical examples, and best practices to help you master this topic.
Define Celery tasks with decorators, configure task options like retries and time limits, bind tasks for introspection, and create custom task classes.
What You Learn
You will learn how to define tasks using the @app.task decorator, configure task options (retries, time limits, rate limits), use bind tasks, create custom task base classes, and organize tasks in modules.
Why It Matters
Task definition is where you configure how Celery executes your code. Properly configured tasks handle failures gracefully, respect system resources, and provide visibility into execution. Poorly defined tasks lead to worker hangs, lost work, and debugging nightmares.
Real-World Use
Doda Browser's analysis pipeline defines tasks with specific time limits, retry policies, and rate limits. File scanning tasks get 5-minute time limits with 3 retries. Notification tasks get 30-second time limits with exponential backoff.
Basic Task Definition
from celery import Celery
app = Celery('tasks', broker='redis://localhost:6379/0')
@app.task
def add(x, y):
"""Add two numbers together."""
return x + y
@app.task
def multiply(x, y):
"""Multiply two numbers."""
return x * y
# client.py
from tasks import add, multiply
print(add.delay(2, 3).get())
print(multiply.delay(4, 5).get())
Expected output:
5
20
Task Options
from celery import Celery
app = Celery('tasks', broker='redis://localhost:6379/0')
# Task with options
@app.task(
bind=True,
max_retries=3,
default_retry_delay=60,
acks_late=True,
time_limit=300,
soft_time_limit=240,
rate_limit='10/m',
ignore_result=True,
serializer='json',
)
def process_file(self, file_path):
"""Process a file with retry on failure."""
try:
with open(file_path, 'r') as f:
content = f.read()
return f"Processed {len(content)} bytes"
except FileNotFoundError as exc:
raise self.retry(exc=exc, countdown=60)
Bind Tasks
The bind=True option makes the task instance accessible as the first argument self:
@app.task(bind=True)
def task_with_state(self, items):
"""Process items and update task state."""
total = len(items)
self.update_state(
state='PROGRESS',
meta={'current': 0, 'total': total}
)
for i, item in enumerate(items):
# Process item
print(f"Processing item {i+1}/{total}: {item}")
self.update_state(
state='PROGRESS',
meta={'current': i + 1, 'total': total}
)
return {'status': 'done', 'total': total}
# client.py
from tasks import task_with_state
result = task_with_state.delay(['a', 'b', 'c', 'd', 'e'])
print(f"Task ID: {result.id}")
# Check progress
print(f"Status: {result.state}")
print(f"Meta: {result.info}")
Expected output:
Task ID: 550e8400-e29b-41d4-a716-446655440000
Status: PENDING
Meta: None
Custom Task Base Class
Create a base class for shared behavior:
from celery import Celery, Task
app = Celery('tasks', broker='redis://localhost:6379/0')
class DatabaseTask(Task):
"""Base task with database connection management."""
abstract = True
def __init__(self):
self.db_connection = None
def before_start(self, task_id, args, kwargs):
self.db_connection = self._connect_db()
print(f"DB connection established for task {task_id}")
def after_return(self, status, retval, task_id, args, kwargs, einfo):
if self.db_connection:
self.db_connection.close()
print(f"DB connection closed for task {task_id}")
def _connect_db(self):
return {"connected": True}
@app.task(base=DatabaseTask, bind=True)
def save_user(self, user_data):
"""Save user data using the database connection."""
result = self.db_connection
print(f"Saving user: {user_data['name']}")
return {"saved": True, "user": user_data['name']}
# client.py
from tasks import save_user
result = save_user.delay({"name": "Alice", "email": "alice@example.com"})
print(result.get(timeout=10))
Expected output:
DB connection established for task 550e8400-...
Saving user: Alice
DB connection closed for task 550e8400-...
{'saved': True, 'user': 'Alice'}
Task Modules Organization
Organize tasks across multiple files:
# tasks/__init__.py
from celery import Celery
app = Celery('project', broker='redis://localhost:6379/0')
# Auto-discover tasks
app.autodiscover_tasks(['tasks.email', 'tasks.files', 'tasks.reports'])
# tasks/email.py
from . import app
@app.task
def send_welcome_email(user_email):
return f"Welcome email sent to {user_email}"
@app.task
def send_password_reset(user_email, token):
return f"Password reset sent to {user_email}"
# tasks/files.py
from . import app
@app.task
def resize_image(image_path, width, height):
return f"Image resized to {width}x{height}"
@app.task
def generate_thumbnail(image_path):
return f"Thumbnail generated for {image_path}"
Task Naming
Celery generates task names from the module path. Control names explicitly:
@app.task(name='email.welcome')
def send_welcome(user_email):
return f"Welcome to {user_email}"
@app.task(name='files.resize')
def resize(image_path, size):
return f"Resized {image_path} to {size}"
Common Mistakes
1. Defining Tasks Inside Functions
Tasks must be importable by the worker. Never define tasks inside request handlers or CLI scripts. Define them in modules that are always importable.
2. Passing Complex Objects as Arguments
Celery serializes task arguments. Pass simple types (strings, numbers, dicts) or use custom serializers. Database model instances are not serializable by default.
3. Forgetting bind=True When Using self
If bind=True is missing, self is not available. The task receives only the positional arguments. Always set bind=True when using task introspection or retry.
4. Not Setting task_serializer
Default serializer is JSON. If you need to pass custom objects, set task_serializer='pickle' but be aware of security implications.
5. Using ignore_result=True When You Need Results
Setting ignore_result=True prevents storing results. Callers trying to call .get() will get None. Only use ignore_result for fire-and-forget tasks.
Practice Questions
1. What does bind=True do in task definition?
It makes the task instance available as the first argument (self), enabling access to task_id, retry(), update_state(), and other task methods.
2. What is the difference between time_limit and soft_time_limit?
soft_time_limit raises a SoftTimeLimitExceeded exception that can be caught. time_limit kills the worker Process. Always set soft_time_limit lower than time_limit.
3. How do you create a custom task base class?
Subclass celery.Task and set abstract = True. Override before_start, after_return, or on_failure to add shared behavior.
4. What is the purpose of autodiscover_tasks?
It automatically discovers and registers tasks in specified modules. Workers can find tasks without importing every module manually.
Challenge
Design a task hierarchy for a document processing system. Create a base task class with logging and metrics tracking. Subclass it for OCR tasks, PDF generation, and text analysis. Each subclass should have appropriate time limits and retry policies.
FAQ
Mini Project: Task Library
# image_tasks.py
from celery import Celery
import base64
import time
app = Celery('image_tasks', broker='redis://localhost:6379/0')
@app.task(bind=True, max_retries=2, acks_late=True, time_limit=60)
def resize_image(self, image_data_b64, width, height):
"""Resize an image (simulated)."""
try:
image_size = len(base64.b64decode(image_data_b64))
print(f"Resizing {image_size} byte image to {width}x{height}")
time.sleep(0.5)
return {
'width': width,
'height': height,
'size': image_size,
'status': 'resized'
}
except Exception as exc:
raise self.retry(exc=exc, countdown=10)
@app.task(bind=True, max_retries=1, time_limit=30)
def generate_thumbnail(self, image_data_b64, size=150):
"""Generate a thumbnail (simulated)."""
image_size = len(base64.b64decode(image_data_b64))
print(f"Generating {size}x{size} thumbnail from {image_size} byte image")
time.sleep(0.3)
return {
'thumbnail_size': size,
'original_size': image_size,
'status': 'thumbnail_generated'
}
@app.task(bind=True, time_limit=120)
def analyze_image(self, image_data_b64):
"""Analyze image content (simulated)."""
image_size = len(base64.b64decode(image_data_b64))
print(f"Analyzing {image_size} byte image")
time.sleep(1.0)
return {
'width': 1920,
'height': 1080,
'format': 'jpeg',
'colors': 16777216,
'status': 'analyzed'
}
# run_pipeline.py
from image_tasks import resize_image, generate_thumbnail, analyze_image
import base64
# Simulate an image
fake_image = base64.b64encode(b'x' * 100000).decode()
# Execute pipeline
resize = resize_image.delay(fake_image, 800, 600)
thumb = generate_thumbnail.delay(fake_image)
analysis = analyze_image.delay(fake_image)
print(resize.get(timeout=10))
print(thumb.get(timeout=10))
print(analysis.get(timeout=10))
Expected output:
Resizing 133333 byte image to 800x600
{'width': 800, 'height': 600, 'size': 133333, 'status': 'resized'}
Generating 150x150 thumbnail from 133333 byte image
{'thumbnail_size': 150, 'original_size': 133333, 'status': 'thumbnail_generated'}
Analyzing 133333 byte image
{'width': 1920, 'height': 1080, 'format': 'jpeg', 'colors': 16777216, 'status': 'analyzed'}
What's Next
Now that you know how to define tasks, learn about running the Celery worker with different concurrency settings, then explore calling tasks with delay and apply_async.
Built by the developers of DodaTech
Doda Browser, DodaZIP & Durga Antivirus Pro