Django-RQ Configuration

Complete guide to configuring Django-RQ in django-cfg projects, including queue setup, scheduling, and advanced Redis configurations.

Quick Start

Minimal Configuration

The simplest Django-RQ configuration uses redis_url from parent DjangoConfig:


# api/config.py
from django_cfg import DjangoConfig
from django_cfg.models import DjangoRQConfig, RQQueueConfig
 
class MyConfig(DjangoConfig):
    # Redis URL (used by Django-RQ automatically)
    redis_url: str = "redis://localhost:6379/0"
 
    # Django-RQ configuration
    django_rq: DjangoRQConfig = DjangoRQConfig(
        enabled=True,
        redis_db=0,  # REQUIRED: Redis database for project isolation (0-15)
        queues=[
            RQQueueConfig(queue="default"),
        ],
    )

:::warning Required Field redis_db is required and must be unique per project to prevent RQ scheduler conflicts. See Redis Isolation for details. :::

That’s it! Django-RQ will automatically:

Use redis_url for all queues
Configure default timeouts and TTL
Register the default queue
Enable admin interface
Add automatic cleanup tasks (daily + weekly)
Add demo heartbeat task (development only)

Complete Configuration

Full Example

Here’s a production-ready configuration with multiple queues and scheduling:


# api/config.py
from django_cfg import DjangoConfig
from django_cfg.models import DjangoRQConfig, RQQueueConfig, RQScheduleConfig
 
class MyConfig(DjangoConfig):
    # === Redis Configuration ===
    redis_url: str = "redis://localhost:6379/0"
 
    # === Django-RQ Configuration ===
    django_rq: DjangoRQConfig = DjangoRQConfig(
        enabled=True,
        redis_db=0,  # REQUIRED: Unique per project (0-15)
 
        # Queue configurations
        queues=[
            # High priority queue for urgent tasks
            RQQueueConfig(
                queue="high",
                default_timeout=180,      # 3 minutes
                default_result_ttl=300,   # 5 minutes
            ),
 
            # Default queue for normal tasks
            RQQueueConfig(
                queue="default",
                default_timeout=360,      # 6 minutes
                default_result_ttl=500,   # 8 minutes
            ),
 
            # Low priority queue for batch operations
            RQQueueConfig(
                queue="low",
                default_timeout=600,      # 10 minutes
                default_result_ttl=800,   # 13 minutes
            ),
 
            # Knowledge base queue (if enable_knowbase=True)
            RQQueueConfig(
                queue="knowledge",
                default_timeout=600,      # 10 minutes
                default_result_ttl=3600,  # 1 hour
            ),
        ],
 
        # Scheduled tasks
        schedules=[
            # Update prices every 5 minutes
            RQScheduleConfig(
                func="apps.crypto.tasks.update_coin_prices",
                interval=300,  # seconds
                queue="default",
                limit=50,
                verbosity=0,
                description="Update coin prices (frequent)",
            ),
 
            # Daily report at midnight
            RQScheduleConfig(
                func="apps.crypto.tasks.generate_report",
                cron="0 0 * * *",  # Cron expression
                queue="low",
                report_type="daily",
                description="Generate daily report",
            ),
        ],
 
        # Admin and monitoring
        show_admin_link=True,
        prometheus_enabled=True,
    )

Configuration Models

DjangoRQConfig

Main configuration model for Django-RQ:

Field	Type	Default	Description
`enabled`	`bool`	`True`	Enable Django-RQ integration
`redis_db`	`int`	required	Redis database number (0-15) for project isolation
`queues`	`List[RQQueueConfig]`	`[RQQueueConfig(queue="default")]`	Queue configurations
`schedules`	`List[RQScheduleConfig]`	`[]`	Scheduled job configurations
`show_admin_link`	`bool`	`True`	Show Django-RQ link in admin
`prometheus_enabled`	`bool`	`True`	Enable Prometheus metrics
`commit_mode`	`str`	`"auto"`	When to enqueue jobs relative to DB transactions (v4+)
`enable_auto_cleanup`	`bool`	`True`	Enable automatic cleanup of old jobs
`cleanup_max_age_days`	`int`	`7`	Maximum age in days before cleanup
`exception_handlers`	`List[str]`	`[]`	Exception handler function paths
`api_token`	`Optional[str]`	`None`	API token for authentication

Methods:


config.to_django_settings(parent_config)  # Generate Django settings
config.get_queue_names()                  # Get list of queue names
config.get_queue_config(name)             # Get specific queue config
config.add_queue(queue_config)            # Add queue programmatically
config.remove_queue(name)                 # Remove queue programmatically
config.get_all_schedules()                # Get all schedules (including auto-generated)

Auto-Generated Schedules:

When enable_auto_cleanup=True (default), Django-RQ automatically adds:

Production & Development:

cleanup_old_jobs - Runs daily (86400s), removes jobs older than cleanup_max_age_days
cleanup_orphaned_job_keys - Runs weekly (604800s), removes orphaned Redis keys

Development Only:

demo_scheduler_heartbeat - Runs every minute (60s), verifies scheduler is working

RQQueueConfig

Configuration for a single queue:

Field	Type	Default	Description
`queue`	`str`	required	Queue name (alphanumeric, hyphens, underscores)
`url`	`Optional[str]`	`None`	Redis URL (overrides host/port/db)
`host`	`str`	`"localhost"`	Redis host
`port`	`int`	`6379`	Redis port
`db`	`int`	`0`	Redis database number (0-15)
`username`	`Optional[str]`	`None`	Redis username (Redis 6+)
`password`	`Optional[str]`	`None`	Redis password
`default_timeout`	`int`	`360`	Default job timeout in seconds
`default_result_ttl`	`int`	`86400`	Default result TTL in seconds (24 hours)
`failure_ttl`	`int`	`604800`	Failed job TTL in seconds (7 days)
`socket_timeout`	`Optional[float]`	`None`	Redis socket timeout
`connection_kwargs`	`Dict[str, Any]`	`{}`	Additional connection arguments
`redis_client_kwargs`	`Dict[str, Any]`	`{}`	Additional Redis client arguments

Advanced Options:


# Redis Sentinel support
RQQueueConfig(
    queue="default",
    sentinels=[("host1", 26379), ("host2", 26379)],
    master_name="mymaster",
    sentinel_kwargs={"password": "sentinel_pass"},
)
 
# SSL/TLS connection
RQQueueConfig(
    queue="default",
    url="rediss://localhost:6380/0",  # Note: rediss://
    connection_kwargs={
        "ssl_cert_reqs": "required",
        "ssl_ca_certs": "/path/to/ca.crt",
    },
)

RQScheduleConfig

Configuration for scheduled jobs:

Field	Type	Default	Description
`func`	`str`	required	Function path (e.g., `apps.myapp.tasks.my_task`)
`cron`	`Optional[str]`	`None`	Cron expression (e.g., `0 0 * * *`)
`interval`	`Optional[int]`	`None`	Interval in seconds
`scheduled_time`	`Optional[str]`	`None`	ISO datetime for one-time job
`queue`	`str`	`"default"`	Queue name
`timeout`	`Optional[int]`	`None`	Job timeout (overrides queue default)
`result_ttl`	`Optional[int]`	`None`	Result TTL (overrides queue default)
`args`	`List[Any]`	`[]`	Positional arguments
`kwargs`	`Dict[str, Any]`	`{}`	Keyword arguments
`job_id`	`Optional[str]`	`None`	Custom job ID
`description`	`Optional[str]`	`None`	Human-readable description
`repeat`	`Optional[int]`	`None`	Number of times to repeat

Declarative Task Parameters:

RQScheduleConfig supports declarative parameters that are automatically added to kwargs:

Parameter	Type	Description
`limit`	`Optional[int]`	Limit parameter for tasks
`verbosity`	`Optional[int]`	Verbosity level (0-3)
`report_type`	`Optional[str]`	Report type parameter
`days`	`Optional[int]`	Days parameter
`force`	`Optional[bool]`	Force parameter
`ignore_errors`	`Optional[bool]`	Continue execution even if task fails

Example:


# Declarative syntax (recommended)
RQScheduleConfig(
    func="apps.crypto.tasks.update_coin_prices",
    interval=300,
    limit=50,          # Automatically added to kwargs
    verbosity=1,       # Automatically added to kwargs
    force=True,        # Automatically added to kwargs
)
 
# Traditional syntax (still works)
RQScheduleConfig(
    func="apps.crypto.tasks.update_coin_prices",
    interval=300,
    kwargs={"limit": 50, "verbosity": 1, "force": True},
)

Queue Configuration

Basic Queue Setup


django_rq: DjangoRQConfig = DjangoRQConfig(
    enabled=True,
    redis_db=0,  # Required: unique per project
    queues=[
        # Single queue with defaults
        RQQueueConfig(queue="default"),
    ],
)

Multiple Queues


django_rq: DjangoRQConfig = DjangoRQConfig(
    enabled=True,
    redis_db=0,  # Required: unique per project
    queues=[
        # High priority (short timeout, short TTL)
        RQQueueConfig(
            queue="high",
            default_timeout=180,
            default_result_ttl=300,
        ),
 
        # Normal priority
        RQQueueConfig(
            queue="default",
            default_timeout=360,
            default_result_ttl=500,
        ),
 
        # Low priority (long timeout, long TTL)
        RQQueueConfig(
            queue="low",
            default_timeout=600,
            default_result_ttl=800,
        ),
    ],
)

Per-Queue Redis Configuration


django_rq: DjangoRQConfig = DjangoRQConfig(
    enabled=True,
    redis_db=0,  # Required: unique per project
    queues=[
        # Queue 1: Use default redis_url with redis_db
        RQQueueConfig(queue="default"),
 
        # Queue 2: Use specific Redis URL (overrides redis_db)
        RQQueueConfig(
            queue="slow",
            url="redis://redis-slow:6379/0",
        ),
 
        # Queue 3: Use different Redis instance
        RQQueueConfig(
            queue="archive",
            host="redis-archive",
            port=6380,
            db=1,
        ),
    ],
)

Queue Naming Best Practices


# Good queue names
"default"        # ✅ Main queue
"high"           # ✅ High priority
"low"            # ✅ Low priority
"email"          # ✅ Specific purpose
"knowledge"      # ✅ AI/knowledge tasks
"reports"        # ✅ Report generation
 
# Bad queue names
"Queue-1"        # ❌ Non-descriptive
"myqueue"        # ❌ Too generic
"QUEUE"          # ❌ All caps
"queue with spaces"  # ❌ Contains spaces (invalid)

Schedule Configuration

Cron Schedules

Use cron expressions for time-based scheduling:


schedules=[
    # Every 5 minutes
    RQScheduleConfig(
        func="apps.myapp.tasks.cleanup",
        cron="*/5 * * * *",
        queue="low",
    ),
 
    # Every day at midnight
    RQScheduleConfig(
        func="apps.myapp.tasks.daily_report",
        cron="0 0 * * *",
        queue="low",
        report_type="daily",
    ),
 
    # Every Monday at 9 AM
    RQScheduleConfig(
        func="apps.myapp.tasks.weekly_summary",
        cron="0 9 * * 1",
        queue="default",
    ),
 
    # First day of month at midnight
    RQScheduleConfig(
        func="apps.myapp.tasks.monthly_invoice",
        cron="0 0 1 * *",
        queue="default",
    ),
]

Cron Expression Format:


* * * * *
│ │ │ │ │
│ │ │ │ └─── Day of week (0-7, 0=Sunday)
│ │ │ └───── Month (1-12)
│ │ └─────── Day of month (1-31)
│ └───────── Hour (0-23)
└─────────── Minute (0-59)

Interval Schedules

Use intervals for periodic tasks:


schedules=[
    # Every 30 seconds
    RQScheduleConfig(
        func="apps.myapp.tasks.health_check",
        interval=30,
        queue="high",
    ),
 
    # Every 5 minutes (300 seconds)
    RQScheduleConfig(
        func="apps.crypto.tasks.update_prices",
        interval=300,
        queue="default",
        limit=100,
    ),
 
    # Every hour (3600 seconds)
    RQScheduleConfig(
        func="apps.crypto.tasks.sync_exchanges",
        interval=3600,
        queue="default",
    ),
 
    # Every 24 hours (86400 seconds)
    RQScheduleConfig(
        func="apps.myapp.tasks.backup",
        interval=86400,
        queue="low",
    ),
]

One-Time Schedules

Schedule jobs for specific times:


from datetime import datetime
 
schedules=[
    # New Year's Eve task
    RQScheduleConfig(
        func="apps.myapp.tasks.new_year_notification",
        scheduled_time="2025-12-31T23:59:00",
        queue="high",
    ),
 
    # Product launch notification
    RQScheduleConfig(
        func="apps.myapp.tasks.launch_notification",
        scheduled_time="2025-06-01T09:00:00",
        queue="high",
    ),
]

Schedule with Task Parameters


schedules=[
    # Update prices with parameters
    RQScheduleConfig(
        func="apps.crypto.tasks.update_coin_prices",
        interval=300,
        queue="default",
        # Declarative parameters (recommended)
        limit=50,
        verbosity=0,
        force=False,
        description="Update top 50 coins every 5 minutes",
    ),
 
    # Generate report with custom parameters
    RQScheduleConfig(
        func="apps.crypto.tasks.generate_report",
        cron="0 0 * * *",
        queue="low",
        # Declarative parameters
        report_type="daily",
        description="Daily market report at midnight",
    ),
 
    # Traditional syntax (still works)
    RQScheduleConfig(
        func="apps.myapp.tasks.custom_task",
        interval=3600,
        args=[],
        kwargs={
            "param1": "value1",
            "param2": 42,
        },
    ),
]

Redis Configuration

Standard Redis


# Using redis_url from parent config (recommended)
class MyConfig(DjangoConfig):
    redis_url: str = "redis://localhost:6379/0"
 
    django_rq: DjangoRQConfig = DjangoRQConfig(
        enabled=True,
        queues=[RQQueueConfig(queue="default")],
    )
 
# Using explicit host/port/db
django_rq: DjangoRQConfig = DjangoRQConfig(
    enabled=True,
    queues=[
        RQQueueConfig(
            queue="default",
            host="localhost",
            port=6379,
            db=0,
        ),
    ],
)

Redis with Authentication


# Redis 6+ with username and password
django_rq: DjangoRQConfig = DjangoRQConfig(
    enabled=True,
    queues=[
        RQQueueConfig(
            queue="default",
            url="redis://username:password@localhost:6379/0",
        ),
    ],
)
 
# Redis < 6 with password only
django_rq: DjangoRQConfig = DjangoRQConfig(
    enabled=True,
    queues=[
        RQQueueConfig(
            queue="default",
            url="redis://:password@localhost:6379/0",
        ),
    ],
)

Redis SSL/TLS


django_rq: DjangoRQConfig = DjangoRQConfig(
    enabled=True,
    queues=[
        RQQueueConfig(
            queue="default",
            url="rediss://localhost:6380/0",  # Note: rediss://
            connection_kwargs={
                "ssl_cert_reqs": "required",
                "ssl_ca_certs": "/path/to/ca.crt",
                "ssl_certfile": "/path/to/client.crt",
                "ssl_keyfile": "/path/to/client.key",
            },
        ),
    ],
)

Redis Sentinel

High availability with Redis Sentinel:


django_rq: DjangoRQConfig = DjangoRQConfig(
    enabled=True,
    queues=[
        RQQueueConfig(
            queue="default",
            # Sentinel configuration
            sentinels=[
                ("sentinel1", 26379),
                ("sentinel2", 26379),
                ("sentinel3", 26379),
            ],
            master_name="mymaster",
            db=0,
            socket_timeout=0.5,
            # Sentinel authentication (if needed)
            sentinel_kwargs={
                "username": "sentinel_user",
                "password": "sentinel_pass",
            },
            # Redis authentication (if needed)
            username="redis_user",
            password="redis_pass",
        ),
    ],
)

Redis Cluster


# Redis Cluster is not directly supported by RQ
# Use a Redis Cluster proxy or stick to standard Redis

Environment-Based Configuration

Development vs Production


from django_cfg import DjangoConfig
from django_cfg.models import DjangoRQConfig, RQQueueConfig
 
class MyConfig(DjangoConfig):
    @property
    def django_rq(self) -> DjangoRQConfig:
        """Environment-aware Django-RQ configuration."""
        if self.is_development:
            # Development: simple setup
            return DjangoRQConfig(
                enabled=True,
                queues=[
                    RQQueueConfig(queue="default"),
                ],
                prometheus_enabled=False,
            )
        else:
            # Production: full setup with multiple queues
            return DjangoRQConfig(
                enabled=True,
                queues=[
                    RQQueueConfig(queue="high", default_timeout=180),
                    RQQueueConfig(queue="default", default_timeout=360),
                    RQQueueConfig(queue="low", default_timeout=600),
                ],
                prometheus_enabled=True,
                schedules=[...],  # Add schedules
            )

YAML-Based Configuration


# config.dev.yaml
redis_url: "redis://localhost:6379/0"
 
django_rq:
  enabled: true
  queues:
    - queue: "default"
      default_timeout: 360
  prometheus_enabled: false
 
# config.prod.yaml
redis_url: "${REDIS_URL}"
 
django_rq:
  enabled: true
  queues:
    - queue: "high"
      default_timeout: 180
    - queue: "default"
      default_timeout: 360
    - queue: "low"
      default_timeout: 600
  prometheus_enabled: true
  schedules:
    - func: "apps.crypto.tasks.update_coin_prices"
      interval: 300
      queue: "default"

Advanced Configuration

Exception Handlers

Custom exception handlers for failed jobs:


# myapp/handlers.py
def log_exception(job, exc_type, exc_value, traceback):
    """Log exception to custom logger."""
    import logging
    logger = logging.getLogger('rq.exceptions')
    logger.error(f"Job {job.id} failed: {exc_value}")
 
def send_exception_to_sentry(job, exc_type, exc_value, traceback):
    """Send exception to Sentry."""
    import sentry_sdk
    sentry_sdk.capture_exception(exc_value)
 
# config.py
django_rq: DjangoRQConfig = DjangoRQConfig(
    enabled=True,
    queues=[...],
    exception_handlers=[
        "myapp.handlers.log_exception",
        "myapp.handlers.send_exception_to_sentry",
    ],
)

Commit Mode (django-rq v4+)

Django-RQ v4 introduced COMMIT_MODE that controls when jobs are enqueued relative to database transactions.


django_rq: DjangoRQConfig = DjangoRQConfig(
    enabled=True,
    redis_db=0,
    commit_mode="auto",  # Default: enqueue immediately (v3-compatible behaviour)
)

Value	Behaviour	When to use
`"auto"`	Enqueue immediately (default)	Compatible with django-rq v3 behaviour
`"on_db_commit"`	Enqueue only after DB transaction commits	Prevents jobs from running if the transaction rolls back
`"request_finished"`	Enqueue at end of HTTP request	Useful for batching multiple enqueues per request

:::tip Recommended Use "on_db_commit" in production to avoid jobs running against data that was never committed to the database. :::

:::note v4 default django-rq v4 ships with COMMIT_MODE = "on_db_commit" as its own default. django-cfg explicitly sets "auto" to preserve backward-compatible v3 behaviour unless you opt in. :::

API Token Authentication

Secure API endpoints with token:


django_rq: DjangoRQConfig = DjangoRQConfig(
    enabled=True,
    queues=[...],
    api_token="your-secret-token-here",
)
 
# Access API with token
# GET /api/cfg/rq/monitor/health/
# Header: Authorization: Token your-secret-token-here

Custom Queue Defaults


# Global timeout for all queues
DEFAULT_TIMEOUT = 360
DEFAULT_RESULT_TTL = 500
 
django_rq: DjangoRQConfig = DjangoRQConfig(
    enabled=True,
    queues=[
        RQQueueConfig(
            queue="default",
            default_timeout=DEFAULT_TIMEOUT,
            default_result_ttl=DEFAULT_RESULT_TTL,
        ),
        RQQueueConfig(
            queue="low",
            default_timeout=DEFAULT_TIMEOUT * 2,
            default_result_ttl=DEFAULT_RESULT_TTL * 2,
        ),
    ],
)

Configuration Validation

Django-cfg automatically validates your configuration:

Validation Rules


# ✅ Valid configurations
RQQueueConfig(queue="default")  # OK
RQQueueConfig(queue="my-queue")  # OK
RQQueueConfig(queue="my_queue")  # OK
 
# ❌ Invalid configurations
RQQueueConfig(queue="")  # Error: queue name required
RQQueueConfig(queue="my queue")  # Error: contains spaces
RQQueueConfig(queue="My Queue!")  # Error: invalid characters
 
# ✅ Valid timeouts
RQQueueConfig(queue="default", default_timeout=360)  # OK
RQQueueConfig(queue="default", default_timeout=1)    # OK
 
# ❌ Invalid timeouts
RQQueueConfig(queue="default", default_timeout=0)    # Error: must be >= 1
RQQueueConfig(queue="default", default_timeout=-1)   # Error: must be >= 1
 
# ✅ Valid schedules
RQScheduleConfig(func="myapp.tasks.task1", interval=60)  # OK
RQScheduleConfig(func="myapp.tasks.task2", cron="0 * * * *")  # OK
 
# ❌ Invalid schedules
RQScheduleConfig(func="myapp.tasks.task3")  # Error: no schedule type
RQScheduleConfig(func="myapp.tasks.task4", interval=60, cron="0 * * * *")  # Error: multiple schedule types

Validation Errors


# Missing required queue
django_rq: DjangoRQConfig = DjangoRQConfig(
    enabled=True,
    queues=[
        RQQueueConfig(queue="high"),
        # Missing: RQQueueConfig(queue="default")
    ],
)
# Error: A queue named 'default' is required
 
# Duplicate queue names
django_rq: DjangoRQConfig = DjangoRQConfig(
    enabled=True,
    queues=[
        RQQueueConfig(queue="default"),
        RQQueueConfig(queue="default"),  # Duplicate
    ],
)
# Error: Duplicate queue names found: {'default'}

Configuration Best Practices

1. Use Multiple Queues


# Good: Separate queues by priority/purpose
queues=[
    RQQueueConfig(queue="high"),     # Critical tasks
    RQQueueConfig(queue="default"),  # Normal tasks
    RQQueueConfig(queue="low"),      # Batch operations
]
 
# Bad: Single queue for everything
queues=[
    RQQueueConfig(queue="default"),
]

2. Set Appropriate Timeouts


# Good: Timeouts match task duration
queues=[
    RQQueueConfig(queue="api_calls", default_timeout=30),     # 30s for API calls
    RQQueueConfig(queue="default", default_timeout=360),      # 6m for normal tasks
    RQQueueConfig(queue="reports", default_timeout=1800),     # 30m for reports
]
 
# Bad: Same timeout for all tasks
queues=[
    RQQueueConfig(queue="default", default_timeout=360),
]

3. Use Declarative Schedule Parameters


# Good: Declarative syntax
RQScheduleConfig(
    func="apps.crypto.tasks.update_prices",
    interval=300,
    limit=50,        # Automatically added to kwargs
    verbosity=1,     # Automatically added to kwargs
)
 
# Bad: Manual kwargs
RQScheduleConfig(
    func="apps.crypto.tasks.update_prices",
    interval=300,
    kwargs={"limit": 50, "verbosity": 1},
)

4. Use Environment Variables


# Good: Environment variables for secrets
class MyConfig(DjangoConfig):
    redis_url: str = os.getenv("REDIS_URL", "redis://localhost:6379/0")
 
# Bad: Hardcoded credentials
class MyConfig(DjangoConfig):
    redis_url: str = "redis://:password@production-redis:6379/0"

5. Enable Monitoring


# Good: Enable monitoring in production
django_rq: DjangoRQConfig = DjangoRQConfig(
    enabled=True,
    show_admin_link=True,
    prometheus_enabled=True,
)
 
# Bad: Disable monitoring
django_rq: DjangoRQConfig = DjangoRQConfig(
    enabled=True,
    show_admin_link=False,
    prometheus_enabled=False,
)

Automatic Cleanup

Overview

Django-RQ includes automatic cleanup enabled by default to keep your Redis instance healthy and prevent memory bloat from accumulated job data.

How It Works

Django-RQ automatically runs three maintenance tasks in the background:

Daily Cleanup (Production & Development):

cleanup_old_jobs - Removes finished and failed jobs older than 7 days
Runs every 24 hours
Keeps recent job history for debugging
Configurable retention period

Weekly Cleanup (Production & Development):

cleanup_orphaned_job_keys - Removes orphaned Redis keys
Runs every 7 days
Cleans up keys left after crashes or improper cancellations

Heartbeat (Development Only):

demo_scheduler_heartbeat - Verifies scheduler is working
Runs every minute
Logs heartbeat messages for monitoring
Automatically disabled in production

Default Configuration

Cleanup is enabled by default with sensible defaults:


django_rq: DjangoRQConfig = DjangoRQConfig(
    enabled=True,
    # Cleanup is enabled by default - no configuration needed!
)

Customize Retention Period

Adjust how long jobs are kept before cleanup:


django_rq: DjangoRQConfig = DjangoRQConfig(
    enabled=True,
    cleanup_max_age_days=14,  # Keep jobs for 2 weeks instead of 7 days
)

Disable Automatic Cleanup

If you prefer manual control:


django_rq: DjangoRQConfig = DjangoRQConfig(
    enabled=True,
    enable_auto_cleanup=False,  # Disable automatic cleanup
)

Manual Cleanup

Run cleanup tasks manually when needed:


from django_cfg.modules.django_rq.tasks.maintenance import (
    cleanup_old_jobs,
    cleanup_orphaned_job_keys,
    get_rq_stats,
)
 
# Clean up jobs older than 7 days
stats = cleanup_old_jobs(max_age_days=7, dry_run=False)
print(f"Deleted {stats['total_deleted']} jobs")
 
# Clean up orphaned keys
stats = cleanup_orphaned_job_keys(dry_run=False)
print(f"Deleted {stats['orphaned_deleted']} keys")
 
# Get RQ statistics
stats = get_rq_stats()
print(f"Queued: {stats['queue']['queued']}")
print(f"Failed: {stats['queue']['failed']}")

Dry Run Mode

Test cleanup without actually deleting anything:


# Preview what would be deleted
stats = cleanup_old_jobs(max_age_days=7, dry_run=True)
print(f"Would delete {stats['total_deleted']} jobs")
 
# Only delete after reviewing
if stats['total_deleted'] < 1000:
    cleanup_old_jobs(max_age_days=7, dry_run=False)

What Gets Cleaned

Removed:

Finished jobs older than cleanup_max_age_days
Failed jobs older than cleanup_max_age_days
Orphaned job keys (weekly)

Protected:

Queued jobs (waiting to be processed)
Running jobs (currently being processed)
Scheduled jobs (cron/interval tasks)
All non-RQ Redis data

Safety Guarantees

Cleanup is designed to be safe for production use:

Isolated scope: Only touches RQ-specific keys (rq:job:*, rq:finished:*, rq:failed:*)
No interference: Scheduled jobs use separate storage and are never cleaned
Data protection: Your application data (custom Redis keys) is never touched
Configurable retention: Keep job history as long as you need
Dry run mode: Test before applying changes

Monitoring Cleanup

Check cleanup task execution in logs:


# View cleanup logs
tail -f logs/djangocfg/integrations.log | grep cleanup
 
# Check scheduled jobs
python manage.py shell -c "
from django_rq import get_scheduler
scheduler = get_scheduler('default')
for job in scheduler.get_jobs():
    if 'cleanup' in job.func_name:
        print(f'{job.func_name}: {job.meta}')
"

Redis Isolation

Why It’s Required

:::danger Critical Issue If multiple projects share the same Redis DB, scheduled tasks will randomly stop working. When any project flushes cache (cache.clear(), FLUSHDB), it deletes ALL keys including RQ scheduler jobs. Tasks execute 2-3 times then disappear forever. :::

When multiple Django projects share a single Redis instance, RQ schedulers can conflict:

Cache Flush Kills Jobs: Any project clearing cache deletes scheduler job metadata
Shared Keys: All schedulers read from the same rq:scheduler:scheduled_jobs set
Job Theft: Workers from one project try to execute jobs from another
Missing Functions: Jobs fail because task functions don’t exist in other projects
Duplicate Execution: Same job runs multiple times across projects

Solution: Separate Redis Databases

Redis supports 16 databases (0-15). Each project uses a unique database number via the required redis_db field:


# Project A: carapis
django_rq: DjangoRQConfig = DjangoRQConfig(
    enabled=True,
    redis_db=0,  # carapis uses DB 0
    queues=[...],
)
 
# Project B: unrealon
django_rq: DjangoRQConfig = DjangoRQConfig(
    enabled=True,
    redis_db=1,  # unrealon uses DB 1
    queues=[...],
)
 
# Project C: myproject
django_rq: DjangoRQConfig = DjangoRQConfig(
    enabled=True,
    redis_db=10,  # myproject uses DB 10
    queues=[...],
)

How It Works

DjangoConfig.redis_url provides the base Redis connection (e.g., redis://localhost:6379/0)
DjangoRQConfig.redis_db replaces the database number in the URL
All queues, workers, and schedulers use the isolated database


# Parent config
redis_url: str = "redis://shared-redis:6379/0"
 
# RQ config
django_rq: DjangoRQConfig = DjangoRQConfig(
    redis_db=5,  # Replaces /0 with /5
    ...
)
 
# Resulting URL: redis://shared-redis:6379/5

Database Assignments

Maintain a central registry of database assignments. Example:

Project	Redis DB	Purpose
carapis	0	Vehicle catalog
unrealon	1	Remote parser management
sdkrouter	2	SDK routing
cmdop	3	AI-powered machine management
listingapis	4	Listing aggregation
propapis	5	Property APIs
reforms	6	Reforms platform
stockapis	7	Stock APIs
telegram	8	Telegram bots
djangocfg	9	Django-CFG demo
available	10-15	Future projects

Validation

The redis_db field is:

Required: No default value - must be explicitly set
Validated: Must be 0-15 (Pydantic validates range)
Documented: Field description explains purpose


# ✅ Valid
DjangoRQConfig(redis_db=5, ...)
 
# ❌ Invalid - missing required field
DjangoRQConfig(queues=[...])  # Error: redis_db field required
 
# ❌ Invalid - out of range
DjangoRQConfig(redis_db=16, ...)  # Error: must be <= 15

Django-RQ Configuration

Quick Start

Minimal Configuration

Complete Configuration

Full Example

Configuration Models

DjangoRQConfig

RQQueueConfig

RQScheduleConfig

Queue Configuration

Basic Queue Setup

Multiple Queues

Per-Queue Redis Configuration

Queue Naming Best Practices

Schedule Configuration

Cron Schedules

Interval Schedules

One-Time Schedules

Schedule with Task Parameters

Redis Configuration

Standard Redis

Redis with Authentication

Redis SSL/TLS

Redis Sentinel

Redis Cluster

Environment-Based Configuration

Development vs Production

YAML-Based Configuration

Advanced Configuration

Exception Handlers

Commit Mode (django-rq v4+)

API Token Authentication

Custom Queue Defaults

Configuration Validation

Validation Rules

Validation Errors

Configuration Best Practices

1. Use Multiple Queues

2. Set Appropriate Timeouts

3. Use Declarative Schedule Parameters

4. Use Environment Variables

5. Enable Monitoring

Automatic Cleanup

Overview

How It Works

Default Configuration

Customize Retention Period

Disable Automatic Cleanup

Manual Cleanup

Dry Run Mode

What Gets Cleaned

Safety Guarantees

Monitoring Cleanup

Redis Isolation

Why It’s Required

Solution: Separate Redis Databases

How It Works

Database Assignments

Validation

See Also

Documentation

Reference

Documentation

Frontend

Resources

Company