Post cover image
Custom Task Scheduling in Python using Redis and Celery

September 21, 2024

Custom Task Scheduling in Python using Redis and Celery

As applications scale, handling asynchronous tasks like sending emails, generating reports, or processing large datasets becomes crucial…

By Pranav Dixit

3 min read

As applications scale, handling asynchronous tasks like sending emails, generating reports, or processing large datasets becomes crucial. Instead of keeping users waiting for these operations, Python provides excellent libraries like Celery for task scheduling and Redis as a broker to manage and store task queues. In this blog, we'll guide you through creating a custom task-scheduling system using Redis and Celery in Python.

What is Celery?

Celery is a distributed task queue. It allows you to run time-consuming operations in the background as asynchronous tasks. Celery handles task scheduling, distribution, and execution, making it an essential tool for creating responsive applications.

Why Redis?

Redis is an in-memory data structure store, widely used as a message broker for Celery. It manages the task queues efficiently, making it a perfect choice for real-time messaging systems.

Setting Up the Environment

Before diving into the code, you need to set up the environment. You'll need:

  1. Python 3.x installed.
  2. Redis server running.
  3. Required Python libraries.
pip install celery redis

Step 1: Install and Configure Redis If you don't already have Redis installed, you can install it by following the official documentation: https://redis.io/docs/getting-started/installation/ After installation, run the Redis server:

redis-server

Step 2: Set Up Celery in a Python Project You can create a new Python project or use an existing one. First, define a Celery instance within your project. 2.1 Install Celery and Redis If you haven't installed Celery yet, run the following:

pip install celery redis

2.2 Create a Celery Configuration Create a celery.py file in your project directory to configure Celery:

from celery import Celery

# Initialize the Celery application
app = Celery(
    'my_app',
    broker='redis://localhost:6379/0',  # Redis URL
    backend='redis://localhost:6379/0'  # Use Redis as the backend
)

# Optional configurations
app.conf.update(
    result_expires=3600,  # Task result expiration time
)

# Load tasks from a module
app.autodiscover_tasks(['tasks'])

In this code, Celery is configured to use Redis as both the broker and the backend to store results.

2.3 Create a Task Module Next, define a tasks module (e.g., tasks.py) where you'll create your background tasks:

from celery import shared_task
import time

@shared_task
def sample_task(x, y):
    # A sample task that takes some time to execute
    time.sleep(5)
    return x + y

This task takes two arguments, x and y, waits for 5 seconds (simulating a time-consuming operation), and then returns their sum.

Step 3: Run Celery Workers Now, you need to run Celery workers to pick up and execute tasks. In your project directory, run the following command:

celery -A celery worker --loglevel=info

This starts with a Celery worker who listens for tasks from the Redis queue. The — loglevel=info flag enables detailed logging of the worker's activity.

Step 4: Call and Schedule Tasks To call a task asynchronously, import the task in your Python script and run it as follows:

from tasks import sample_task

# Call the task asynchronously
result = sample_task.delay(4, 6)

# Get the result (will block until the task completes)
print(f"Task Result: {result.get()}")

The .delay() method sends the task to the queue, and result.get() retrieves the task's result once it's done. However, this blocks the execution until the result is ready, so use it only when necessary.

Step 5: Scheduling Tasks Periodically To schedule tasks to run periodically, you can use Celery's built-in periodic task scheduler, Celery Beat. This requires additional configuration. First, install the additional dependencies:

pip install celery[redis] celery[redis_scheduler]

Then, configure Celery Beat in your celery.py file:

from celery.schedules import crontab

app.conf.beat_schedule = {
    'run-sample-task-every-30-seconds': {
        'task': 'tasks.sample_task',
        'schedule': 30.0,  # Run every 30 seconds
        'args': (2, 3),
    },
}

In this configuration, the sample_task is scheduled to run every 30 seconds with arguments (2, 3).

Step 6: Run Celery Beat To start the scheduler, run Celery Beat alongside the workers in a separate terminal:

celery -A celery beat --loglevel=info

Now, the sample_task will execute every 30 seconds.

Step 7: Handling Task Results You can store and manage the results of your tasks using Redis as the backend. Once tasks are completed, results can be retrieved as shown in Step 4, but Celery also provides more advanced result management using AsyncResult.

from celery.result import AsyncResult

# Retrieve a result by task ID
task_id = result.id  # Replace with the task ID
async_result = AsyncResult(task_id)

if async_result.successful():
    print(f"Task Result: {async_result.result}")

This allows you to check the status of tasks and handle results asynchronously.

Conclusion

By integrating Celery and Redis into your Python project, you can efficiently schedule and manage asynchronous tasks. Whether you're executing tasks in the background or scheduling them periodically, this setup allows for scalable and maintainable task execution. With a solid foundation of Celery and Redis, you can now focus on customizing your task scheduling according to your application's needs. Happy coding!