Django RQ cheatsheet

django_rq_cs.py

import django_rq
import rq
import rq.registry

from conf import settings


conn = django_rq.get_connection()

# Get all workers
workers = django_rq.get_worker().all(con)
for worker in workers:
  print(worker.hostname)

# Get all queues
queues = []
for i in range(len(settings.RQ_QUEUES)):
  queues.append(django_rq.queues.get_queue_by_index(i))

# Access jobs registries
q_default = queues[0]
scheduled_job_registry = rq.registry.ScheduledJobRegistry(q_default.name, conn)
failed_job_registry = rq.registry.FailedJobRegistry(q_default.name, conn)
finished_job_registry = rq.registry.FinishedJobRegistry(q_default.name, conn)

# Get IDs of jobs in a registry
job_ids = failed_job_registry.get_job_ids()

# Fetch a job from its ID
job0 = rq.job.Job.fetch(job_ids[0], conn)

# Interesting infos of a job
job0.created_at
job0.enqueued_at
job0.ended_at
job0.func_name