tessl/pypi-django-rq

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

—

Pending

Overview

Eval results

Files

Job Management

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

Core functionality for creating, enqueueing, and managing background jobs in Django-RQ. This includes job decoration, direct enqueueing, and job lifecycle management.

Capabilities

Job Decorator

Transform regular Python functions into RQ tasks using the @job decorator.

def job(func_or_queue, connection=None, *args, **kwargs):
    """
    Decorator to create RQ tasks from functions.
    
    Args:
        func_or_queue: Function to decorate or queue name/instance
        connection: Redis connection (optional)
        *args, **kwargs: RQ job options (timeout, result_ttl, etc.)
    
    Returns:
        Decorated function with .delay() and .enqueue() methods
    """

Usage examples:

from django_rq import job

# Default queue
@job
def simple_task():
    return "Task completed"

# Specific queue
@job('high')
def priority_task():
    return "High priority task"

# With options
@job('default', timeout=3600)
def long_task():
    return "Long running task"

# Execute jobs
simple_task.delay()
priority_task.delay()
long_task.delay()

Direct Job Enqueueing

Enqueue functions directly without decoration.

def enqueue(func, *args, **kwargs):
    """
    Enqueue a job to the default queue.
    
    Args:
        func: Function to execute
        *args: Positional arguments for func
        **kwargs: Keyword arguments for func
    
    Returns:
        Job: RQ Job instance
    """

Usage example:

import django_rq

def my_function(arg1, arg2):
    return arg1 + arg2

# Enqueue to default queue
job = django_rq.enqueue(my_function, 10, 20)
print(f"Job ID: {job.id}")

Worker Management

Create and configure RQ workers for processing jobs.

def get_worker(*queue_names, job_class=None, queue_class=None, worker_class=None, **kwargs):
    """
    Create an RQ worker for specified queues.
    
    Args:
        *queue_names: Names of queues to process
        job_class: Custom job class (optional)
        queue_class: Custom queue class (optional) 
        worker_class: Custom worker class (optional)
        **kwargs: Additional worker options
    
    Returns:
        Worker: RQ Worker instance
    """

def get_worker_class(worker_class=None):
    """
    Get worker class from settings or return default.
    
    Args:
        worker_class: Override worker class
        
    Returns:
        type: Worker class
    """

def get_exception_handlers():
    """
    Get custom exception handlers from settings.
    
    Returns:
        list: Exception handler functions
    """

Usage examples:

import django_rq

# Worker for default queue
worker = django_rq.get_worker()
worker.work(burst=True)  # Process all jobs then exit

# Worker for multiple queues
worker = django_rq.get_worker('high', 'default', 'low')
worker.work()  # Run continuously

# Worker with custom classes
from myapp.workers import CustomWorker
worker = django_rq.get_worker('default', worker_class=CustomWorker)

Job Class Configuration

Configure custom job classes for specialized job handling.

def get_job_class(job_class=None):
    """
    Get job class from settings or return default.
    
    Args:
        job_class: Override job class (class or import path)
        
    Returns:
        type: Job class
    """