An app that provides django integration for RQ (Redis Queue)
npx @tessl/cli install tessl/pypi-django-rq@3.1.0Django integration for RQ (Redis Queue), a Python library for queueing jobs and processing them in the background with workers. Django-RQ provides seamless integration with Django's settings system, management commands for running workers, a web-based dashboard for monitoring queue status, and optional integrations with monitoring tools.
pip install django-rqimport django_rqFor job decoration:
from django_rq import jobFor queue and worker management:
from django_rq import get_queue, get_worker, get_connection, get_schedulerimport django_rq
from django_rq import job
# Basic job enqueueing
def my_task(arg1, arg2):
return arg1 + arg2
# Enqueue job to default queue
django_rq.enqueue(my_task, 10, 20)
# Using @job decorator
@job
def decorated_task():
return "Hello from RQ!"
# Queue job using decorator
decorated_task.delay()
# Working with specific queues
queue = django_rq.get_queue('high')
queue.enqueue(my_task, 5, 15)
# Creating workers
worker = django_rq.get_worker('default', 'high')
worker.work(burst=True)Django-RQ extends RQ's core components with Django-specific features:
Core functionality for creating, enqueueing, and managing background jobs including job decoration, queue operations, and worker management.
def enqueue(func, *args, **kwargs):
"""Enqueue a job to the default queue."""
@job
def decorated_function():
"""Decorator to create RQ tasks."""
def get_queue(name='default', **kwargs):
"""Get a queue instance by name."""
def get_worker(*queue_names, **kwargs):
"""Get a worker instance for specified queues."""Queue creation, configuration, and management including multiple queue support, Redis connection handling, and queue statistics.
def get_connection(name='default', use_strict_redis=False):
"""Get Redis connection for a queue."""
def get_queues(*queue_names, **kwargs):
"""Get multiple queue instances."""
class DjangoRQ:
"""Django-specific Queue class with autocommit support."""
def enqueue_call(self, *args, **kwargs): ...Job scheduling functionality including delayed execution, periodic tasks, and scheduler management.
def get_scheduler(name='default', **kwargs):
"""Get scheduler instance for delayed/periodic jobs."""
class DjangoScheduler:
"""Django-specific Scheduler class."""
def _create_job(self, *args, **kwargs): ...Django management commands for worker control, queue monitoring, and system maintenance.
# Management commands (via manage.py)
# python manage.py rqworker [queues...]
# python manage.py rqscheduler
# python manage.py rqstats
# python manage.py rqsuspend
# python manage.py rqresumeWeb-based monitoring interface for queues, jobs, workers, and statistics with Django admin integration.
# URL patterns for web interface
urlpatterns = [
path('django-rq/', include('django_rq.urls'))
]Prometheus metrics collection, Sentry integration, and Django-specific monitoring features.
from django_rq.utils import get_statistics, configure_sentry
from django_rq.contrib.prometheus import RQCollectorDjango-RQ is configured through Django settings:
# settings.py
RQ_QUEUES = {
'default': {
'HOST': 'localhost',
'PORT': 6379,
'DB': 0,
},
'high': {
'HOST': 'localhost',
'PORT': 6379,
'DB': 0,
}
}
# Optional settings
RQ = {
'DEFAULT_RESULT_TTL': 500,
'AUTOCOMMIT': True,
}