tessl/pypi-django-rq

An app that provides django integration for RQ (Redis Queue)

—

Pending

Overview

Eval results

Files

Queue Operations

Name: tessl/pypi-django-rq
Author: tessl

Queue creation, configuration, and management functionality including multiple queue support, Redis connection handling, and queue statistics.

Capabilities

Queue Access

Get queue instances configured through Django settings.

def get_queue(name='default', default_timeout=None, is_async=None, autocommit=None, 
              connection=None, queue_class=None, job_class=None, serializer=None, **kwargs):
    """
    Get an RQ Queue instance using RQ_QUEUES configuration.
    
    Args:
        name: Queue name from RQ_QUEUES (default: 'default')
        default_timeout: Override default job timeout
        is_async: Whether queue runs asynchronously
        autocommit: Whether jobs are committed immediately
        connection: Redis connection instance
        queue_class: Custom queue class
        job_class: Custom job class
        serializer: Custom serializer
        **kwargs: Additional queue options
        
    Returns:
        DjangoRQ: Queue instance
    """

def get_queues(*queue_names, **kwargs):
    """
    Get multiple queue instances that share the same Redis connection.
    
    Args:
        *queue_names: Queue names to retrieve
        **kwargs: Options passed to get_queue()
        
    Returns:
        list: List of queue instances
        
    Raises:
        ValueError: If queues have different classes or connections
    """

Usage examples:

import django_rq

# Get default queue
queue = django_rq.get_queue()

# Get named queue
high_queue = django_rq.get_queue('high')

# Get queue with custom options
queue = django_rq.get_queue('default', 
                           default_timeout=600,
                           is_async=True)

# Get multiple queues
queues = django_rq.get_queues('high', 'default', 'low')

Redis Connection Management

Manage Redis connections for queues.

def get_connection(name='default', use_strict_redis=False):
    """
    Get Redis connection for a named queue.
    
    Args:
        name: Queue name from RQ_QUEUES
        use_strict_redis: Use StrictRedis class
        
    Returns:
        redis.Redis: Redis connection instance
    """

def get_redis_connection(config, use_strict_redis=False):
    """
    Create Redis connection from configuration.
    
    Args:
        config: Queue configuration dict
        use_strict_redis: Use StrictRedis class
        
    Returns:
        redis.Redis: Redis connection instance
    """

def get_unique_connection_configs(config=None):
    """
    Get unique Redis connection configurations.
    
    Args:
        config: Queue configurations (uses RQ_QUEUES if None)
        
    Returns:
        list: Unique connection configurations
    """

Usage examples:

import django_rq

# Get connection for queue
conn = django_rq.get_connection('high')

# Share connection across queues
conn = django_rq.get_connection('default')
queue1 = django_rq.get_queue('queue1', connection=conn)
queue2 = django_rq.get_queue('queue2', connection=conn)

Django-Specific Queue Class

Enhanced queue class with Django integration features.

class DjangoRQ:
    """
    Django-specific Queue class extending RQ's Queue.
    
    Features:
    - Autocommit support for Django transactions
    - Settings integration for result TTL
    - Request/response cycle job enqueueing
    """
    
    def __init__(self, *args, autocommit=None, **kwargs):
        """
        Initialize Django RQ queue.
        
        Args:
            *args: Positional arguments for Queue
            autocommit: Override autocommit setting
            **kwargs: Keyword arguments for Queue
        """
    
    def enqueue_call(self, *args, **kwargs):
        """
        Enqueue job with Django transaction support.
        
        Respects autocommit setting - jobs may be delayed
        until end of Django request if autocommit=False.
        """
    
    def original_enqueue_call(self, *args, **kwargs):
        """
        Direct enqueue bypassing autocommit behavior.
        
        Always enqueues immediately regardless of autocommit setting.
        """

Queue Configuration

Configure queue classes and behavior.

def get_queue_class(config=None, queue_class=None):
    """
    Get queue class from configuration or settings.
    
    Args:
        config: Queue configuration dict
        queue_class: Override queue class
        
    Returns:
        type: Queue class to use
    """

def get_commit_mode():
    """
    Get autocommit mode from RQ settings.
    
    Returns:
        bool: Whether jobs are committed immediately
    """

Queue Statistics and Monitoring

from django_rq.utils import get_statistics

def get_statistics(run_maintenance_tasks=False):
    """
    Get comprehensive queue statistics.
    
    Note: This function is in the utils module and must be imported directly.
    
    Args:
        run_maintenance_tasks: Whether to run cleanup tasks
        
    Returns:
        dict: Statistics including job counts, worker info, etc.
    """

Statistics include:

Queue job counts by status (queued, started, finished, failed, deferred, scheduled)
Worker counts and details
Oldest job timestamps
Connection information
Scheduler status

Supported Redis Configurations

Django-RQ supports various Redis configurations:

Basic Connection

RQ_QUEUES = {
    'default': {
        'HOST': 'localhost',
        'PORT': 6379,
        'DB': 0,
    }
}

URL-based Connection

RQ_QUEUES = {
    'default': {
        'URL': 'redis://localhost:6379/0',
    }
}

Sentinel Configuration

RQ_QUEUES = {
    'default': {
        'SENTINELS': [('localhost', 26379)],
        'MASTER_NAME': 'mymaster',
        'DB': 0,
    }
}

Django Cache Integration

RQ_QUEUES = {
    'default': {
        'USE_REDIS_CACHE': 'redis-cache',
    }
}

SSL Connection

RQ_QUEUES = {
    'default': {
        'HOST': 'localhost',
        'PORT': 6380,
        'DB': 0,
        'SSL': True,
        'SSL_CERT_REQS': 'required',
    }
}