RQ: Simple Job Queues for Python with Redis/Valkey

Introduction

RQ (Redis Queue) is a simple yet powerful Python library for managing job queues and processing tasks asynchronously in the background. Backed by Redis or Valkey, RQ is designed for ease of use, allowing developers to quickly integrate background processing into their applications. It offers a robust solution for projects ranging from small utilities to large-scale enterprise systems, providing excellent scalability and reliability.

Installation

Getting started with RQ is straightforward. You can install the latest version using pip:

$ pip install rq

RQ requires a running Redis or Valkey server (version 5 or higher for Redis, 7.2 or higher for Valkey).

Examples

Basic Job Enqueueing

To enqueue a job, first define your function, then create an RQ queue and add the function call to it:

import requests
from redis import Redis
from rq import Queue

def count_words_at_url(url):
    """Just an example function that's called async."""
    resp = requests.get(url)
    return len(resp.text.split())

# Create an RQ queue
queue = Queue(connection=Redis())

# Enqueue the function call
job = queue.enqueue(count_words_at_url, 'https://stamps.id')
print(f"Job ID: {job.id}")

Job Prioritization

RQ allows you to prioritize jobs in two ways:

1. Enqueue at the front:

   job = queue.enqueue(count_words_at_url, 'https://stamps.id', at_front=True)
   

2. Use multiple queues: Define separate queues for different priorities and start workers with a prioritized list.

   from rq import Queue
   high_priority_queue = Queue('high', connection=Redis())
   low_priority_queue = Queue('low', connection=Redis())
   
   high_priority_queue.enqueue(urgent_task)
   low_priority_queue.enqueue(non_urgent_task)
   

   Then, start a worker:

   $ rq worker high low
   

Scheduling Jobs

Schedule jobs to run at a specific time or after a delay:

from datetime import datetime, timedelta
# Schedule job to run at 9:15, October 10th
job = queue.enqueue_at(datetime(2019, 10, 10, 9, 15), say_hello)

# Schedule job to run in 10 seconds
job = queue.enqueue_in(timedelta(seconds=10), say_hello)

Repeating Jobs

Execute a job multiple times using the Repeat class:

from rq import Repeat

# Repeat job 3 times after successful execution, with 30 second intervals
queue.enqueue(my_function, repeat=Repeat(times=3, interval=30))

Unique Jobs

Prevent duplicate jobs from being enqueued:

job = queue.enqueue(send_email, user_id, job_id='welcome-42', unique=True)

Rate Limiting

Apply concurrency-based rate limits to jobs sharing a key:

from rq import RateLimit

queue.enqueue(generate_report, rate_limit=RateLimit(key='reports', concurrency=2))

Retrying Failed Jobs

Configure jobs to retry upon failure:

from rq import Retry

# Retry up to 3 times, failed job will be requeued immediately
queue.enqueue(say_hello, retry=Retry(max=3))

Webhooks

Send HTTP requests to a URL when a job finishes or fails:

from rq import Webhook

queue.enqueue(
    say_hello,
    webhooks=[
        Webhook('https://example.com/finished', job_status='finished'),
        Webhook('https://example.com/failed', job_status='failed', method='POST'),
    ],
)

Cron Job Scheduling

RQ provides built-in functionality for interval-based and cron syntax scheduling. Define your jobs in a configuration file:

# cron_config.py
from rq import cron
from myapp import cleanup_temp_files, generate_analytics_report

cron.register(
    cleanup_temp_files,
    queue_name='maintenance',
    interval=1800  # 30 minutes in seconds
)

cron.register(
    generate_analytics_report,
    queue_name='reports',
    cron='0 8 1 * *' # Monthly report on the first day of each month at 8:00 AM
)

Then, start the rq cron command:

$ rq cron cron_config.py

The Worker

To process enqueued jobs, start an RQ worker:

$ rq worker --with-scheduler

For production, you can use rq worker-pool to run multiple worker processes:

$ rq worker-pool -n 4

Why Use RQ?

RQ stands out for its simplicity and Pythonic approach to background job processing. It offers a lightweight alternative to more complex systems, making it easy to get started while providing powerful features like job prioritization, scheduling, retries, and webhooks. Its reliance on Redis/Valkey ensures high performance and reliability, making it suitable for a wide range of asynchronous tasks, from sending emails to complex data processing.

Links

• Official Documentation: python-rq.org
• GitHub Repository: github.com/rq/rq