Calling Celery Tasks — Complete Guide
In this tutorial, you will learn about Calling Celery Tasks. We cover key concepts, practical examples, and best practices to help you master this topic.
Call Celery tasks using delay() and apply_async(), control execution with countdown, eta, and priority, and handle task results and timeouts.
What You Learn
You will learn the difference between delay() and apply_async(), how to schedule tasks with countdown and eta, set priorities, pass custom options, and handle task results and timeouts.
Why It Matters
How you call tasks determines when and where they execute. Using the right calling method gives you control over task timing, routing, and prioritization. The wrong approach leads to performance issues and unexpected behavior.
Real-World Use
Doda Browser uses apply_async for malware analysis tasks with countdown for retries, eta for scheduled scans, and priority to ensure critical threats are analyzed before routine scans.
delay vs apply_async
delay() is a shortcut for apply_async() with minimal options. apply_async() provides full control.
from celery import Celery
app = Celery('calling', broker='redis://localhost:6379/0')
@app.task
def process(data):
return f"Processed: {data}"
# delay - simple and fast
result1 = process.delay("simple call")
# apply_async - full control
result2 = process.apply_async(
args=("full control",),
kwargs={},
countdown=10,
expires=3600,
priority=5,
queue='high_priority',
routing_key='high.process',
task_id='custom-id-123',
retry=True,
retry_policy={
'max_retries': 3,
'interval_start': 0,
'interval_step': 0.2,
'interval_max': 0.2,
},
)
print(f"Task 1: {result1.id}")
print(f"Task 2: {result2.id}")
Expected output:
Task 1: 550e8400-e29b-41d4-a716-446655440000
Task 2: 550e8400-e29b-41d4-a716-446655440001
Scheduling with Countdown and ETA
from celery import Celery
from datetime import datetime, timedelta
app = Celery('schedule', broker='redis://localhost:6379/0')
@app.task
def scheduled_task(name):
return f"Executed: {name}"
# Execute after 30 seconds
result1 = scheduled_task.apply_async(
args=("countdown",),
countdown=30
)
# Execute at specific time
eta = datetime.utcnow() + timedelta(hours=2)
result2 = scheduled_task.apply_async(
args=("eta",),
eta=eta.isoformat()
)
# Execute after 1 hour with expiry
result3 = scheduled_task.apply_async(
args=("expires",),
countdown=3600,
expires=7200
)
print(f"Countdown task: {result1.id}, scheduled in 30s")
print(f"ETA task: {result2.id}, scheduled at {eta}")
print(f"Expires task: {result3.id}, valid for 2 hours")
Expected output:
Countdown task: 550e8400..., scheduled in 30s
ETA task: 6ba7b810..., scheduled at 2026-06-28 12:00:00
Expires task: 6ba7b811..., valid for 2 hours
Task Priority
from celery import Celery
app = Celery('priority', broker='redis://localhost:6379/0')
# Configure queue with priorities
app.conf.update(
task_queue_max_priority=10,
task_default_priority=5,
)
@app.task
def priority_task(name, level):
return f"{name} (priority {level}) processed"
# Submit tasks with different priorities
tasks = []
for priority in range(10):
task = priority_task.apply_async(
args=(f"task_{priority}", priority),
priority=priority
)
tasks.append(task)
for t in tasks:
print(f"{t.id[:8]}: {t.get(timeout=10)}")
Expected output:
550e8400: task_0 (priority 0) processed
550e8401: task_1 (priority 1) processed
...
550e8409: task_9 (priority 9) processed
Passing Arguments
from celery import Celery
app = Celery('args', broker='redis://localhost:6379/0')
@app.task
def process_args(*args, **kwargs):
return {
'args': args,
'kwargs': kwargs,
'length_args': len(args),
'length_kwargs': len(kwargs),
}
# Positional arguments
result1 = process_args.delay(1, 2, 3)
# Keyword arguments
result2 = process_args.delay(a=1, b=2, c=3)
# Mixed
result3 = process_args.delay(1, 2, key='value')
print(f"Positional: {result1.get()}")
print(f"Keyword: {result2.get()}")
print(f"Mixed: {result3.get()}")
Expected output:
Positional: {'args': (1, 2, 3), 'kwargs': {}, 'length_args': 3, 'length_kwargs': 0}
Keyword: {'args': (), 'kwargs': {'a': 1, 'b': 2, 'c': 3}, 'length_args': 0, 'length_kwargs': 3}
Mixed: {'args': (1, 2), 'kwargs': {'key': 'value'}, 'length_args': 2, 'length_kwargs': 1}
Handling Results
from celery import Celery
import time
app = Celery('results', broker='redis://localhost:6379/0')
@app.task(bind=True, track_started=True)
def long_task(self, duration):
for i in range(duration):
time.sleep(1)
self.update_state(
state='PROGRESS',
meta={'current': i + 1, 'total': duration}
)
return "done"
# Client code
result = long_task.delay(10)
# Check status
print(f"Status: {result.state}")
print(f"Task ID: {result.id}")
# Wait with timeout
try:
output = result.get(timeout=15)
print(f"Result: {output}")
except TimeoutError:
print("Task did not finish in time")
# Check without blocking
print(f"Ready: {result.ready()}")
print(f"Success: {result.successful()}")
print(f"Failed: {result.failed()}")
if result.ready():
print(f"Result value: {result.result}")
Expected output:
Status: PENDING
Task ID: 550e8400-...
[after 10 seconds]
Result: done
Ready: True
Success: True
Failed: False
Result value: done
Common Mistakes
1. Blocking on Every Task
Calling .get() on every task defeats the purpose of async processing. Use .get() only when you need the result. For fire-and-forget tasks, ignore the result.
2. Using delay for Complex Scheduling
delay() accepts only positional arguments. Use apply_async() when you need countdown, eta, priority, or custom routing.
3. Not Setting Timeouts
Without timeouts, a hung task blocks .get() forever. Always set timeout in .get() and configure task_time_limit on the task.
4. Passing Non-Serializable Arguments
Celery serializes arguments to JSON by default. Passing database models, file handles, or datetime objects causes Serialization errors. Convert to primitives first.
5. Ignoring Task.expires
Tasks can stay in the queue indefinitely. Set expires for time-sensitive tasks like password reset emails that should not be sent after a deadline.
Practice Questions
1. What is the difference between delay() and apply_async()?
delay() is a shortcut that only accepts positional args. apply_async() accepts args, kwargs, and all execution options (countdown, eta, priority, queue, etc.).
2. How do you schedule a task to run in 1 hour?
Use apply_async(countdown=3600) or eta=datetime.utcnow() + timedelta(hours=1). countdown is seconds; eta is an absolute datetime.
3. What happens when task expires passes?
The task is discarded if it has not started executing by the expires time. The broker removes it from the queue.
4. How do you set task priority?
Use apply_async(priority=N) where N is 0-9. Also configure task_queue_max_priority=10 and task_default_priority=5 in Celery config.
Challenge
Build a task scheduling system for a social media platform. Tasks include: post publication (on-time), notification dispatch (5-minute delay), report generation (off-peak hours), and cleanup (low priority, expires in 24h). Use appropriate calling methods for each.
FAQ
Mini Project: Task Caller
from celery import Celery
from datetime import datetime, timedelta
import time
app = Celery('caller_demo', broker='redis://localhost:6379/0')
@app.task(bind=True, max_retries=3)
def process_order(self, order_id, items, priority='normal'):
print(f"Processing order {order_id}: {len(items)} items")
print(f"Priority: {priority}")
print(f"Processing...")
time.sleep(1)
return {
'order_id': order_id,
'status': 'processed',
'items_count': len(items),
'processed_at': datetime.utcnow().isoformat()
}
# Urgent order - process immediately
urgent = process_order.apply_async(
args=("ORD-001", ["item1", "item2"]),
kwargs={'priority': 'high'},
priority=9,
queue='urgent'
)
print(f"Urgent: {urgent.id}")
# Standard order with delay
standard = process_order.apply_async(
args=("ORD-002", ["item3", "item4", "item5"]),
countdown=10,
priority=5
)
print(f"Standard (10s delay): {standard.id}")
# Scheduled order
scheduled = process_order.apply_async(
args=("ORD-003", ["item6"]),
eta=(datetime.utcnow() + timedelta(hours=1)).isoformat(),
expires=(datetime.utcnow() + timedelta(hours=2)).isoformat()
)
print(f"Scheduled (1 hour): {scheduled.id}")
# Check urgent result
print(f"\nUrgent result: {urgent.get(timeout=10)}")
Expected output:
Processing order ORD-001: 2 items
Priority: high
Processing...
Urgent result: {'order_id': 'ORD-001', 'status': 'processed', 'items_count': 2, 'processed_at': '2026-06-28T10:00:00'}
Standard (10s delay): 6ba7b810-...
Scheduled (1 hour): 6ba7b811-...
What's Next
Now that you know how to call tasks, learn about task retry and error handling for building robust task pipelines, then explore task routing for directing tasks to specific workers.
Built by the developers of DodaTech
Doda Browser, DodaZIP & Durga Antivirus Pro