tessl/pypi-django-rq

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

—

Pending

Overview

Eval results

Files

Scheduling

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

Job scheduling functionality for delayed execution, periodic tasks, and scheduler management using RQ's built-in scheduler or the optional rq-scheduler package.

Capabilities

Scheduler Access

Get scheduler instances for managing delayed and periodic jobs.

def get_scheduler(name='default', queue=None, interval=60, connection=None):
    """
    Get RQ Scheduler instance for delayed/periodic jobs.
    
    Args:
        name: Queue name for scheduler
        queue: Queue instance (optional)
        interval: Scheduler polling interval in seconds
        connection: Redis connection (optional)
        
    Returns:
        DjangoScheduler: Scheduler instance
        
    Raises:
        ImproperlyConfigured: If rq-scheduler not installed
    """

Usage examples:

import django_rq
from datetime import datetime, timedelta

# Get scheduler for default queue
scheduler = django_rq.get_scheduler()

# Get scheduler for specific queue
scheduler = django_rq.get_scheduler('high')

# Schedule job for specific time
def my_task():
    return "Scheduled task executed"

# Schedule for specific datetime
run_time = datetime.now() + timedelta(hours=1)
job = scheduler.enqueue_at(run_time, my_task)

# Schedule with delay
job = scheduler.enqueue_in(timedelta(minutes=30), my_task)

Built-in RQ Scheduler

RQ 1.2.0+ includes built-in scheduling support.

# Using queue methods for built-in scheduler
queue = django_rq.get_queue('default')

# Schedule job for specific time
from datetime import datetime
run_time = datetime(2024, 12, 25, 9, 0, 0)
job = queue.enqueue_at(run_time, my_task)

# Schedule with delay
from datetime import timedelta
job = queue.enqueue_in(timedelta(hours=2), my_task)

Django-Specific Scheduler

Enhanced scheduler class with Django settings integration.

class DjangoScheduler:
    """
    Django-specific Scheduler extending rq-scheduler.
    
    Features:
    - Automatic timeout configuration from RQ_QUEUES
    - Result TTL configuration from RQ settings
    - Integration with Django queue configuration
    """
    
    def _create_job(self, *args, **kwargs):
        """
        Create scheduled job with Django settings integration.
        
        Automatically applies:
        - DEFAULT_TIMEOUT from queue configuration
        - DEFAULT_RESULT_TTL from RQ settings
        """

Scheduler Management Commands

Django management commands for scheduler control.

# Management commands (via manage.py)
# python manage.py rqscheduler [--interval=60] [--queue=default]

The rqscheduler command:

Starts the RQ scheduler daemon
Processes scheduled jobs at configured intervals
Supports custom scheduler classes via settings
Handles graceful shutdown and error recovery

Periodic Jobs

Create recurring jobs with various scheduling patterns.

import django_rq
from datetime import datetime, timedelta

scheduler = django_rq.get_scheduler()

def maintenance_task():
    return "Maintenance completed"

# Schedule recurring job
from rq_scheduler import Scheduler

# Daily at 2 AM
scheduler.schedule(
    scheduled_time=datetime.now().replace(hour=2, minute=0, second=0),
    func=maintenance_task,
    interval=86400,  # 24 hours in seconds
    repeat=None      # Repeat indefinitely
)

# Weekly backup
scheduler.schedule(
    scheduled_time=datetime.now() + timedelta(days=7),
    func=maintenance_task,
    interval=604800,  # 1 week in seconds
    repeat=10        # Repeat 10 times
)

Cron-style Scheduling

Using the rq-cron package for cron-like scheduling:

from rq_cron import cron

# Register cron job
@cron('0 2 * * *')  # Daily at 2 AM
def daily_cleanup():
    return "Daily cleanup completed"

# Register with specific queue
@cron('0 */6 * * *', queue='maintenance')  # Every 6 hours
def periodic_maintenance():
    return "Maintenance completed"

Scheduler Configuration

Configure scheduler behavior through Django settings:

# settings.py
RQ = {
    'SCHEDULER_CLASS': 'myapp.schedulers.CustomScheduler',
    'DEFAULT_RESULT_TTL': 3600,
}

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

Scheduler Statistics

Monitor scheduled jobs and scheduler status:

def get_scheduler_statistics():
    """
    Get scheduler statistics across all Redis connections.
    
    Returns:
        dict: Scheduler statistics including job counts
    """

Statistics include:

Number of scheduled jobs per connection
Scheduler status and configuration
Connection information for each scheduler

Working with Scheduled Jobs

Job Management

import django_rq

scheduler = django_rq.get_scheduler()

# List scheduled jobs
jobs = scheduler.get_jobs()

# Cancel scheduled job
scheduler.cancel(job)

# Reschedule job
scheduler.cancel(job)
new_job = scheduler.enqueue_at(new_time, job.func, *job.args, **job.kwargs)