Celery, the distributed task queue, doesn’t actually run tasks; it orchestrates the scheduling and execution of tasks across one or more workers, which are separate processes.
Let’s see it in action. Imagine you have a Flask app that needs to send a lot of emails. Doing this synchronously would block your web server, making your app unresponsive.
# app.py
from flask import Flask, request, jsonify
from tasks import send_welcome_email # We'll define this next
app = Flask(__name__)
@app.route('/signup', methods=['POST'])
def signup():
user_data = request.json
user_email = user_data.get('email')
user_name = user_data.get('name')
if not user_email or not user_name:
return jsonify({"error": "Missing email or name"}), 400
# Instead of sending the email here, we send it to Celery
send_welcome_email.delay(user_email, user_name)
return jsonify({"message": "Signup successful! Welcome email will be sent shortly."}), 200
if __name__ == '__main__':
app.run(debug=True)
Now, our tasks.py file contains the actual email sending logic, decorated to be a Celery task:
# tasks.py
from celery import Celery
import time # For demonstration
# Configure Celery
# The first argument is the name of the current module
# The second argument is the broker URL. Redis is common.
# The result_backend is where Celery stores task results.
celery_app = Celery('tasks', broker='redis://localhost:6379/0', backend='redis://localhost:6379/0')
@celery_app.task
def send_welcome_email(email, name):
print(f"Sending welcome email to {name} at {email}...")
# Simulate sending email (e.g., using smtplib)
time.sleep(5) # This would normally be I/O bound
print(f"Successfully sent welcome email to {email}.")
return f"Email sent to {email}"
To make this work, you need three components running:
- The Flask Application: Your web server.
- The Message Broker: Redis (or RabbitMQ, etc.). This is where Flask puts tasks and Celery workers pick them up.
- The Celery Worker: A separate process that monitors the message broker, picks up tasks, and executes them.
You’d start these like so:
- Redis:
redis-server(if you have it installed and in your PATH) - Celery Worker: In your terminal, navigate to your project directory and run:
celery -A tasks worker --loglevel=info(-A taskstells Celery to look for tasks in thetasks.pyfile). - Flask App:
python app.py
When a POST request hits /signup on your Flask app, send_welcome_email.delay(user_email, user_name) is called. delay is a shortcut for apply_async. This tells Celery to serialize the task name (send_welcome_email) and its arguments (user_email, user_name) and send them to the Redis broker. The Celery worker, listening to Redis, sees this new message, deserializes it, and executes the send_welcome_email function with the provided arguments. The time.sleep(5) would happen in the background, not blocking your Flask server.
The core problem Celery solves is separating long-running or I/O-bound operations from your main application thread. This keeps your web requests fast and your application responsive. Your Flask app becomes an API endpoint that enqueues work, and your Celery workers are the doers of that work.
The broker is the communication channel. Think of it as a shared mailbox. Flask puts messages (tasks) into the mailbox, and Celery workers read messages from it. The backend is where task results and status are stored. This is useful for checking if a task succeeded, failed, or is still running.
The most surprising thing about Celery is how little it does itself. It’s a dispatcher, a scheduler, a status reporter – but the actual computation is done by independent worker processes. You could have thousands of Celery workers, all running the same code, picking up tasks from a single broker, and Celery handles the distribution.
The celery_app.task decorator is what registers your Python function as a Celery task. When you call send_welcome_email.delay(), you’re not directly calling the function. Instead, you’re sending a message to Celery instructing it to execute that function on a worker. The worker then fetches the task, deserializes the arguments, and executes the actual Python code.
You can control how tasks are executed. For example, to ensure a task runs only once, you can use bind=True and check state, or more simply, use the acks_late=True option with retries. Another useful pattern is bind=True on the task decorator, which passes the self argument (the task instance) to your task function. This allows you to access task metadata like self.request.id or call self.retry() for automatic retries on failure.
# tasks.py - revised with bind=True and retry
from celery import Celery
import time
celery_app = Celery('tasks', broker='redis://localhost:6379/0', backend='redis://localhost:6379/0')
@celery_app.task(bind=True)
def send_welcome_email(self, email, name):
try:
print(f"Attempt {self.request.retries + 1}: Sending welcome email to {name} at {email}...")
# Simulate a transient failure
if self.request.retries < 2:
raise ConnectionError("Simulated network issue")
time.sleep(5)
print(f"Successfully sent welcome email to {email}.")
return f"Email sent to {email}"
except Exception as e:
print(f"Failed to send email to {email}: {e}")
# Retry the task up to 3 times, with a 60-second delay between retries
raise self.retry(exc=e, countdown=60, max_retries=3)
To trigger this retry behavior, you’d need to run the worker with acks_late=True to ensure the task isn’t marked as done until it finishes successfully.
The next concept you’ll likely encounter is managing task results and handling failures gracefully, perhaps by using callbacks or chains of tasks.