Repository files navigation

Django integration withRQ, aRedisbased Python queuing library.Django-RQ is asimple app that allows you to configure your queues in django'ssettings.pyand easily use them in your project.

If you finddjango-rq useful, please consider supporting its development viaTidelift.

• Django (3.2+)
• RQ

• Installdjango-rq (ordownload from PyPI):

pipinstalldjango-rq

• Adddjango_rq toINSTALLED_APPS insettings.py:

INSTALLED_APPS= (# other apps"django_rq",)

• Configure your queues in django'ssettings.py:

RQ_QUEUES= {'default': {'HOST':'localhost','PORT':6379,'DB':0,'USERNAME':'some-user','PASSWORD':'some-password','DEFAULT_TIMEOUT':360,'DEFAULT_RESULT_TTL':800,'REDIS_CLIENT_KWARGS': {# Eventual additional Redis connection arguments'ssl_cert_reqs':None,        },    },'with-sentinel': {'SENTINELS': [('localhost',26736), ('localhost',26737)],'MASTER_NAME':'redismaster','DB':0,# Redis username/password'USERNAME':'redis-user','PASSWORD':'secret','SOCKET_TIMEOUT':0.3,'CONNECTION_KWARGS': {# Eventual additional Redis connection arguments'ssl':True        },'SENTINEL_KWARGS': {# Eventual Sentinel connection arguments# If Sentinel also has auth, username/password can be passed here'username':'sentinel-user','password':'secret',        },    },'high': {'URL':os.getenv('REDISTOGO_URL','redis://localhost:6379/0'),# If you're on Heroku'DEFAULT_TIMEOUT':500,    },'low': {'HOST':'localhost','PORT':6379,'DB':0,    }}RQ_EXCEPTION_HANDLERS= ['path.to.my.handler']# If you need custom exception handlers

• Includedjango_rq.urls in yoururls.py:

urlpatterns+= [path('django-rq/',include('django_rq.urls'))]

Django-RQ allows you to easily put jobs into any of the queues defined insettings.py. It comes with a few utility functions:

• enqueue - push a job to thedefault queue:

importdjango_rqdjango_rq.enqueue(func,foo,bar=baz)

• get_queue - returns anQueue instance.

importdjango_rqqueue=django_rq.get_queue('high')queue.enqueue(func,foo,bar=baz)

In addition toname argument,get_queue also acceptsdefault_timeout,is_async,autocommit,connection andqueue_class arguments. For example:

queue=django_rq.get_queue('default',autocommit=True,is_async=True,default_timeout=360)queue.enqueue(func,foo,bar=baz)

You can provide your own singleton Redis connection object to this function so that it will notcreate a new connection object for each queue definition. This will help you limitnumber of connections to Redis server. For example:

importdjango_rqimportredisredis_cursor=redis.StrictRedis(host='',port='',db='',password='')high_queue=django_rq.get_queue('high',connection=redis_cursor)low_queue=django_rq.get_queue('low',connection=redis_cursor)

• get_connection - accepts a single queue name argument (defaults to "default")and returns a connection to the queue's Redis server:

importdjango_rqredis_conn=django_rq.get_connection('high')

• get_worker - accepts optional queue names and returns a new RQWorker instance for specified queues (ordefault queue):

importdjango_rqworker=django_rq.get_worker()# Returns a worker for "default" queueworker.work()worker=django_rq.get_worker('low','high')# Returns a worker for "low" and "high"

To easily turn a callable into an RQ task, you can also use the@jobdecorator that comes withdjango_rq:

fromdjango_rqimportjob@jobdeflong_running_func():passlong_running_func.delay()# Enqueue function in "default" queue@job('high')deflong_running_func():passlong_running_func.delay()# Enqueue function in "high" queue

You can pass in any arguments that RQ's job decorator accepts:

@job('default',timeout=3600)deflong_running_func():passlong_running_func.delay()# Enqueue function with a timeout of 3600 seconds.

It's possible to specify default forresult_ttl decorator keyword argumentviaDEFAULT_RESULT_TTL setting:

RQ= {'DEFAULT_RESULT_TTL':5000,}

With this setting, job decorator will setresult_ttl to 5000 unless it'sspecified explicitly.

django_rq provides a management command that starts a worker for every queuespecified as arguments:

python manage.py rqworker high default low

If you want to runrqworker in burst mode, you can pass in the--burst flag:

python manage.py rqworker high default low --burst

If you need to use custom worker, job or queue classes, it is best to use global settings(seeCustom queue classes andCustom job and worker classes). However, it is also possibleto override such settings with command line options as follows.

To use a custom worker class, you can pass in the--worker-class flagwith the path to your worker:

python manage.py rqworker high default low --worker-class 'path.to.GeventWorker'

To use a custom queue class, you can pass in the--queue-class flagwith the path to your queue class:

python manage.py rqworker high default low --queue-class 'path.to.CustomQueue'

To use a custom job class, provide--job-class flag.

Starting from version 2.10, running RQ's worker-pool is also supported:

python manage.py rqworker-pool default low medium --num-workers 4

With RQ 1.2.0. you can usebuilt-in schedulerfor your jobs. For example:

fromdjango_rq.queuesimportget_queuequeue=get_queue('default')job=queue.enqueue_at(datetime(2020,10,10),func)

If you are using built-in scheduler you have to start workers with scheduler support:

python manage.py rqworker --with-scheduler

Alternatively you can useRQ Scheduler.After install you can also use theget_scheduler function to return aScheduler instance for queues defined in settings.py'sRQ_QUEUES.For example:

importdjango_rqscheduler=django_rq.get_scheduler('default')job=scheduler.enqueue_at(datetime(2020,10,10),func)

You can also use the management commandrqscheduler to start the scheduler:

python manage.py rqscheduler

If you havedjango-redis ordjango-redis-cacheinstalled, you can instruct django_rq to use the same connection informationfrom your Redis cache. This has two advantages: it's DRY and it takes advantageof any optimization that may be going on in your cache setup (like usingconnection pooling orHiredis.)

To use configure it, use a dict with the keyUSE_REDIS_CACHE pointing to thename of the desired cache in yourRQ_QUEUES dict. It goes without sayingthat the chosen cache must exist and use the Redis backend. See your respectiveRedis cache package docs for configuration instructions. It's also important topoint out that since the django-redis-cacheShardedClient splits the cacheover multiple Redis connections, it does not work.

Here is an example settings fragment for django-redis:

CACHES= {'redis-cache': {'BACKEND':'redis_cache.cache.RedisCache','LOCATION':'localhost:6379:1','OPTIONS': {'CLIENT_CLASS':'django_redis.client.DefaultClient','MAX_ENTRIES':5000,        },    },}RQ_QUEUES= {'high': {'USE_REDIS_CACHE':'redis-cache',    },'low': {'USE_REDIS_CACHE':'redis-cache',    },}

Sometimes you may want to suspend RQ to prevent it from processing new jobs.A classic example is during the initial phase of a deployment script or in advanceof putting your site into maintenance mode. This is particularly helpful whenyou have jobs that are relatively long-running and might otherwise be forciblykilled during the deploy.

The suspend command stops workers on _all_ queues (in a single Redis database)from picking up new jobs. However currently running jobs will continue untilcompletion.

# Suspend indefinitelypython manage.py rqsuspend# Suspend for a specific duration (in seconds) then automatically# resume work again.python manage.py rqsuspend -d 600# Resume work again.python manage.py rqresume

django_rq also provides a dashboard to monitor the status of your queues at/django-rq/ (or whatever URL you set in yoururls.py during installation.

You can also add a link to this dashboard link in/admin by addingRQ_SHOW_ADMIN_LINK = True insettings.py. Be careful though, this willoverride the default admin template so it may interfere with other apps thatmodifies the default admin template.

These statistics are also available in JSON format via/django-rq/stats.json, which is accessible to staff members.If you need to access this view via otherHTTP clients (for monitoring purposes), you can defineRQ_API_TOKEN and access it via/django-rq/stats.json/<API_TOKEN>.

Note: Statistics of scheduled jobs display jobs fromRQ built-in scheduler,not optionalRQ scheduler.

Additionally, these statistics are also accessible from the command line.

python manage.py rqstatspython manage.py rqstats --interval=1# Refreshes every secondpython manage.py rqstats --json# Output as JSONpython manage.py rqstats --yaml# Output as YAML

Sentryshould be configured within the Djangosettings.py as described in theSentry docs.

You can override the default Django Sentry configuration when running therqworker commandby passing thesentry-dsn option:

./manage.py rqworker --sentry-dsn=https://*****@sentry.io/222222

This will override any existing Django configuration and reinitialise Sentry,setting the following Sentry options:

{'debug':options.get('sentry_debug'),'ca_certs':options.get('sentry_ca_certs'),'integrations': [RedisIntegration(),RqIntegration(),DjangoIntegration()]}

RQ uses Python'slogging, this means you can easily configurerqworker's logging mechanism in django'ssettings.py. For example:

LOGGING= {"version":1,"disable_existing_loggers":False,"formatters": {"rq_console": {"format":"%(asctime)s %(message)s","datefmt":"%H:%M:%S",        },    },"handlers": {"rq_console": {"level":"DEBUG","class":"rq.logutils.ColorizingStreamHandler","formatter":"rq_console","exclude": ["%(asctime)s"],        },    },'loggers': {"rq.worker": {"handlers": ["rq_console","sentry"],"level":"DEBUG"        },    }}

By default, every queue will useDjangoRQ class. If you want to use a custom queue class, you can do soby adding aQUEUE_CLASS option on a per queue basis inRQ_QUEUES:

RQ_QUEUES= {'default': {'HOST':'localhost','PORT':6379,'DB':0,'QUEUE_CLASS':'module.path.CustomClass',    }}

or you can specifyDjangoRQ to use a custom class for all your queues inRQ settings:

RQ= {'QUEUE_CLASS':'module.path.CustomClass',}

Custom queue classes should inherit fromdjango_rq.queues.DjangoRQ.

If you are using more than one queue class (not recommended), be sure to only run workerson queues with same queue class. For example if you have two queues defined inRQ_QUEUES andone has custom class specified, you would have to run at least two separate workers for eachqueue.

Similarly to custom queue classes, global custom job and worker classes can be configured usingJOB_CLASS andWORKER_CLASS settings:

RQ= {'JOB_CLASS':'module.path.CustomJobClass','WORKER_CLASS':'module.path.CustomWorkerClass',}

Custom job class should inherit fromrq.job.Job. It will be used for all jobsif configured.

Custom worker class should inherit fromrq.worker.Worker. It will be used for runningall workers unless overridden byrqworker management commandworker-class option.

For an easier testing process, you can run a worker synchronously this way:

fromdjango.testimportTestCasefromdjango_rqimportget_workerclassMyTest(TestCase):deftest_something_that_creates_jobs(self):        ...# Stuff that init jobs.get_worker().work(burst=True)# Processes all jobs then stop.        ...# Asserts that the job stuff is done.

You can set the optionASYNC toFalse to make synchronous operation thedefault for a given queue. This will cause jobs to execute immediately and onthe same thread as they are dispatched, which is useful for testing anddebugging. For example, you might add the following after you queueconfiguration in your settings file:

# ... Logic to set DEBUG and TESTING settings to True or False ...# ... Regular RQ_QUEUES setup code ...ifDEBUGorTESTING:forqueueConfiginRQ_QUEUES.values():queueConfig['ASYNC']=False

Note that setting theis_async parameter explicitly when callingget_queuewill override this setting.

To rundjango_rq's test suite:

`which django-admin` test django_rq --settings=django_rq.tests.settings --pythonpath=.

Create an rqworker service that runs the high, default, and low queues.

sudo vi /etc/systemd/system/rqworker.service

[Unit]Description=Django-RQ WorkerAfter=network.target[Service]WorkingDirectory=<<path_to_your_project_folder>>ExecStart=/home/ubuntu/.virtualenv/<<your_virtualenv>>/bin/python\    <<path_to_your_project_folder>>/manage.py\    rqworker high default low[Install]WantedBy=multi-user.target

Enable and start the service

sudo systemctlenable rqworkersudo systemctl start rqworker

Add django-rq to your requirements.txt file with:

pip freeze> requirements.txt

Update your Procfile to:

web: gunicorn --pythonpath="$PWD/your_app_name" config.wsgi:applicationworker: python your_app_name/manage.py rqworker high default low

Commit and re-deploy. Then add your new worker with:

heroku scale worker=1

SeeCHANGELOG.md.