Python Celery for Distributed Task Queues
Celery is a simple, flexible, and reliable distributed system to process vast amounts of messages, while providing operations with the tools required to maintain such a system. It is a task queue with focus on real-time processing, while also supporting task scheduling.
Installation
Celery requires a message broker to send and receive messages. RabbitMQ and Redis are the most common choices.
pip install celery
# For Redis support
pip install "celery[redis]"Basic Application
Create a file named tasks.py.
from celery import Celery
# Initialize Celery
# Broker URL format: transport://userid:password@hostname:port/virtual_host
app = Celery('tasks', broker='redis://localhost:6379/0', backend='redis://localhost:6379/0')
@app.task
def add(x, y):
return x + yRunning the Worker
You can start the worker by executing the program with the worker argument.
celery -A tasks worker --loglevel=INFO-A tasks: The name of the module (file) containing the app instance.worker: Starts the worker process.--loglevel=INFO: Sets the logging level.
Calling Tasks
To call a task, use the delay() method. This is a shortcut for apply_async().
from tasks import add
# This sends the task to the broker and returns an AsyncResult instance
result = add.delay(4, 4)
# Check if the task is ready
print(f"Ready: {result.ready()}")
# Get the result (blocks until ready)
try:
print(f"Result: {result.get(timeout=1)}")
except Exception as e:
print(f"Task failed: {e}")Configuration
Celery can be configured via a configuration object or a separate file.
app.conf.update(
task_serializer='json',
accept_content=['json'], # Ignore other content
result_serializer='json',
timezone='UTC',
enable_utc=True,
)Periodic Tasks (Celery Beat)
Celery Beat is a scheduler that kicks off tasks at regular intervals.
celery -A tasks beat