RQ

https://python-rq.org/docs
Admin
RQ is a simple Python library for queueing jobs and processing them in the background with workers,...
Tokens:21,986
Snippets:264
Trust Score:9.4
Update:2 months ago
Show doc for...
Tokens:
Raw
### Install and Run RQ Dashboard

Source: https://python-rq.org/docs/monitoring

Install the rq-dashboard package and launch the web-based monitoring interface.

```bash
$ pip install rq-dashboard
$ rq-dashboard
```

--------------------------------

### CLI Enqueueing Examples

Source: https://python-rq.org/docs

Examples demonstrating how CLI arguments map to Python queue.enqueue calls using different parsing methods.

```bash
rq enqueue path.to.func abc -> queue.enqueue(path.to.func, 'abc')
rq enqueue path.to.func abc=def -> queue.enqueue(path.to.func, abc='def')
rq enqueue path.to.func ':{"json": "abc"}' -> queue.enqueue(path.to.func, {'json': 'abc'})
rq enqueue path.to.func 'key:={"json": "abc"}' -> queue.enqueue(path.to.func, key={'json': 'abc'})
rq enqueue path.to.func '%1, 2' -> queue.enqueue(path.to.func, (1, 2))
rq enqueue path.to.func '%None' -> queue.enqueue(path.to.func, None)
rq enqueue path.to.func '%True' -> queue.enqueue(path.to.func, True)
rq enqueue path.to.func 'key%=(1, 2)' -> queue.enqueue(path.to.func, key=(1, 2))
rq enqueue path.to.func 'key%={"foo": True}' -> queue.enqueue(path.to.func, key={"foo": True})
rq enqueue path.to.func @path/to/file -> queue.enqueue(path.to.func, open('path/to/file', 'r').read())
rq enqueue path.to.func key=@path/to/file -> queue.enqueue(path.to.func, key=open('path/to/file', 'r').read())
rq enqueue path.to.func :@path/to/file.json -> queue.enqueue(path.to.func, json.loads(open('path/to/file.json', 'r').read()))
rq enqueue path.to.func key:=@path/to/file.json -> queue.enqueue(path.to.func, key=json.loads(open('path/to/file.json', 'r').read()))
```

--------------------------------

### Install setproctitle for Process Titles

Source: https://python-rq.org/docs/workers

Install the setproctitle package to allow RQ workers to set descriptive process titles visible in system monitoring tools.

```bash
pip install setproctitle
```

--------------------------------

### Start the RQ scheduler

Source: https://python-rq.org/docs/cron

Execute the scheduler process using the command line interface pointing to the configuration file.

```bash
rq cron cron_config.py
```

--------------------------------

### Start an RQ Worker

Source: https://python-rq.org/docs/workers

Run a worker from the project root to process jobs from specified queues.

```bash
$ rq worker high default low
*** Listening for work on high, default, low
Got send_newsletter('me@nvie.com') from default
Job ended normally without result
*** Listening for work on high, default, low
...
```

--------------------------------

### Run RQ worker with scheduler

Source: https://python-rq.org/docs/scheduling

Command to start an RQ worker with the scheduler component enabled.

```bash
$ rq worker --with-scheduler
```

--------------------------------

### Run RQ worker with scheduler programmatically

Source: https://python-rq.org/docs/scheduling

Initialize and start a worker with the scheduler enabled using the Python API.

```python
from rq import Worker, Queue
from redis import Redis

redis = Redis()
queue = Queue(connection=redis)

worker = Worker(queues=[queue], connection=redis)
worker.work(with_scheduler=True)
```

--------------------------------

### Start Worker with Serializer via CLI

Source: https://python-rq.org/docs/workers

Use the --serializer flag in the command line to specify the serializer for the worker process.

```bash
$ rq worker --serializer json
```

--------------------------------

### Push and pop connections manually

Source: https://python-rq.org/docs/connections

Use push_connection and pop_connection methods when a context manager is not suitable, such as in unit testing setups.

```python
import unittest
from rq import Queue
from rq import push_connection, pop_connection

class MyTest(unittest.TestCase):
    def setUp(self):
        push_connection(Redis())

    def tearDown(self):
        pop_connection()

    def test_foo(self):
        """Any queues created here use local Redis."""
        q = Queue()
```

--------------------------------

### Implement a custom worker script with preloading

Source: https://python-rq.org/docs/workers

Use a custom script to preload heavy modules before the worker starts its loop to improve throughput.

```python
#!/usr/bin/env python
from redis import Redis
from rq import Worker

# Preload libraries
import library_that_you_want_preloaded

# Provide the worker with the list of queues (str) to listen to.
w = Worker(['default'], connection=Redis())
w.work()
```

--------------------------------

### Run Workers with Custom Serializer

Source: https://python-rq.org/docs/jobs

Specify the serializer when starting workers via the command line to match the queue configuration.

```bash
$ rq worker --serializer json
$ rq worker --serializer rq.serializers.JSONSerializer
```

--------------------------------

### Start a worker with a custom name

Source: https://python-rq.org/docs/workers

Override the default randomly generated worker name by providing a name argument during instantiation.

```python
from redis import Redis
from rq import Queue, Worker

redis = Redis()
queue = Queue('queue_name')

# Start a worker with a custom name
worker = Worker([queue], connection=redis, name='foo')
```

--------------------------------

### Configuring Job Retries

Source: https://python-rq.org/docs/exceptions

Configure automatic retries for failed jobs using the Retry object. Ensure workers are started with --with-scheduler when using intervals.

```python
from redis import Redis
from rq import Retry, Queue

from somewhere import my_func

queue = Queue(connection=redis)
# Retry up to 3 times, failed job will be requeued immediately
queue.enqueue(my_func, retry=Retry(max=3))

# Retry up to 3 times, with 60 seconds interval in between executions
queue.enqueue(my_func, retry=Retry(max=3, interval=60))

# Retry up to 3 times, with longer interval in between retries
queue.enqueue(my_func, retry=Retry(max=3, interval=[10, 30, 60]))
```

--------------------------------

### Accessing and querying a StartedJobRegistry

Source: https://python-rq.org/docs/job_registries

Demonstrates how to instantiate a StartedJobRegistry and perform basic operations like counting jobs, listing IDs, and checking for job existence.

```python
import time
from redis import Redis
from rq import Queue
from rq.registry import StartedJobRegistry
from somewhere import count_words_at_url

redis = Redis()
queue = Queue(connection=redis)
job = queue.enqueue(count_words_at_url, 'http://nvie.com')

# get StartedJobRegistry by queue
registry = StartedJobRegistry(queue=queue)

# or get StartedJobRegistry by queue name and connection
registry2 = StartedJobRegistry(name='my_queue', connection=redis)

# sleep for a moment while job is taken off the queue
time.sleep(0.1)

print('Queue associated with the registry: %s' % registry.get_queue())
print('Number of jobs in registry %s' % registry.count)

# get the list of ids for the jobs in the registry
print('IDs in registry %s' % registry.get_job_ids())

# test if a job is in the registry using the job instance or job id
print('Job in registry %s' % (job in registry))
print('Job in registry %s' % (job.id in registry))
```

--------------------------------

### Run RQ Worker with configuration file

Source: https://python-rq.org/docs/workers

Use the -c flag to specify the module containing worker settings.

```bash
$ rq worker -c settings
```

--------------------------------

### Manage RQ Queues and Jobs

Source: https://python-rq.org/docs

Demonstrates common queue operations such as counting jobs, fetching job instances, and clearing queue data.

```python
from rq import Queue
from redis import Redis

redis_conn = Redis()
q = Queue(connection=redis_conn)

# Getting the number of jobs in the queue
# Note: Only queued jobs are counted, not including deferred ones
print(len(q))

# Retrieving jobs
queued_job_ids = q.job_ids # Gets a list of job IDs from the queue
queued_jobs = q.jobs # Gets a list of enqueued job instances
job = q.fetch_job('my_id') # Returns job having ID "my_id"

# Emptying a queue, this will delete all jobs in this queue
q.empty()

# Deleting a queue
q.delete(delete_jobs=True) # Passing in `True` will remove all jobs in the queue
# queue is now unusable. It can be recreated by enqueueing jobs to it.
```

--------------------------------

### Execute jobs synchronously for testing

Source: https://python-rq.org/docs

Initialize a Queue with is_async=False to execute jobs immediately in the current process. A Redis connection is still required for state management.

```python
>>> q = Queue('low', is_async=False, connection=my_redis_conn)
>>> job = q.enqueue(fib, 8)
>>> job.result
21
```

--------------------------------

### Configure RQ Worker with settings.py

Source: https://python-rq.org/docs/workers

Define Redis connection, queues, Sentry DSN, and logging configuration in a Python file.

```python
REDIS_URL = 'redis://localhost:6379/1'

# You can also specify the Redis DB to use
# REDIS_HOST = 'redis.example.com'
# REDIS_PORT = 6380
# REDIS_DB = 3
# REDIS_PASSWORD = 'very secret'

# Queues to listen on
QUEUES = ['high', 'default', 'low']

# If you're using Sentry to collect your runtime exceptions, you can use this
# to configure RQ for it in a single step
# The 'sync+' prefix is required for raven: https://github.com/nvie/rq/issues/350#issuecomment-43592410
SENTRY_DSN = 'sync+http://public:secret@example.com/1'

# If you want custom worker name
# NAME = 'worker-1024'

# If you want to use a dictConfig <https://docs.python.org/3/library/logging.config.html#logging-config-dictschema>
# for more complex/consistent logging requirements.
DICT_CONFIG = {
    'version': 1,
    'disable_existing_loggers': False,
    'formatters': {
        'standard': {
            'format': '%(asctime)s [%(levelname)s] %(name)s: %(message)s'
        },
    },
    'handlers': {
        'default': {
            'level': 'INFO',
            'formatter': 'standard',
            'class': 'logging.StreamHandler',
            'stream': 'ext://sys.stderr',  # Default is stderr
        },
    },
    'loggers': {
        'root': {  # root logger
            'handlers': ['default'],
            'level': 'INFO',
            'propagate': False
        },
    }
}
```

--------------------------------

### Configuring Immediate Repetition

Source: https://python-rq.org/docs/scheduling

Shows how to set an interval of zero to trigger immediate re-enqueueing upon job completion.

```python
# This job will repeat 3 times with no delay between executions
job = queue.enqueue(my_function, repeat=Repeat(times=3, interval=0))
```

--------------------------------

### Pass connection via constructor

Source: https://python-rq.org/docs/connections

The recommended approach for handling connections is to pass a Redis instance to the constructor of RQ objects.

```python
from redis import Redis
from rq import Queue

redis = Redis('localhost', 6379)
q = Queue(connection=redis)
```

```python
from rq import Queue
from redis import Redis

conn1 = Redis('localhost', 6379)
conn2 = Redis('remote.host.org', 9836)

q1 = Queue('foo', connection=conn1)
q2 = Queue('bar', connection=conn2)
```

--------------------------------

### Get Job Position in Queue

Source: https://python-rq.org/docs/jobs

Retrieves the current position of a job within the queue. Note that this operation iterates over all jobs and may perform poorly on large queues.

```python
from rq import Queue
from redis import Redis
from hello import say_hello

redis_conn = Redis()
q = Queue(connection=redis_conn)

job = q.enqueue(say_hello)
job2 = q.enqueue(say_hello)

job2.get_position()
# returns 1

q.get_job_position(job)
# return 0
```

--------------------------------

### Enqueue a job for background execution

Source: https://python-rq.org/docs

Initialize a queue with a Redis connection and enqueue a function call to be processed by a worker.

```python
from rq import Queue
from redis import Redis
from somewhere import count_words_at_url
import time

# Tell RQ what Redis/Valkey connection to use
redis_conn = Redis()
q = Queue(connection=redis_conn)  # no args implies the default queue

# Delay execution of count_words_at_url('http://nvie.com')
job = q.enqueue(count_words_at_url, 'http://nvie.com')
print(job.result)   # => None  # Changed to job.return_value() in RQ >= 1.12.0

# Now, wait a while, until the worker is finished
time.sleep(2)
print(job.result)   # => 889  # Changed to job.return_value() in RQ >= 1.12.0
```

--------------------------------

### Queue.prepare_data(func, args=None, job_id=None, ...)

Source: https://python-rq.org/docs

Prepares job data for bulk enqueueing. Accepts the same arguments as standard queueing methods.

```APIDOC
## Queue.prepare_data(func, args=None, job_id=None)

### Description
Prepares the necessary data structure for a job to be enqueued via `enqueue_many`.

### Parameters
- **func** (callable) - Required - The function to be executed.
- **args** (tuple) - Optional - Arguments to pass to the function.
- **job_id** (str) - Optional - A unique identifier for the job.
```

--------------------------------

### Create a job instance manually

Source: https://python-rq.org/docs/jobs

Use Job.create to instantiate a job object before adding it to a queue with enqueue_job.

```python
from rq.job import Job

job = Job.create(count_words_at_url, 'http://nvie.com')
print('Job id: %s' % job.id)
q.enqueue_job(job)

# create a job with a predetermined id
job = Job.create(count_words_at url, 'http://nvie.com', id='my_job_id')
```

--------------------------------

### Basic CLI Enqueue Command

Source: https://python-rq.org/docs

The standard syntax for enqueueing a job via the command line.

```bash
rq enqueue [OPTIONS] FUNCTION [ARGUMENTS]
```

--------------------------------

### Configure worker with custom classes

Source: https://python-rq.org/docs/workers

Use command-line arguments to specify custom job and queue classes for the worker process.

```bash
$ rq worker --job-class 'custom.JobClass' --queue-class 'custom.QueueClass'
```

--------------------------------

### Enqueueing Repeating Jobs

Source: https://python-rq.org/docs/scheduling

Demonstrates basic usage of the Repeat class with fixed and variable intervals.

```python
from redis import Redis
from rq import Queue, Repeat

queue = Queue(connection=Redis())

# Repeat a job 3 times with the same interval of 30 seconds
job = queue.enqueue(my_function, repeat=Repeat(times=3, interval=30))

# Or use different intervals between repetitions
job = queue.enqueue(my_function, repeat=Repeat(times=3, interval=[5, 10, 15]))
```

--------------------------------

### rq enqueue

Source: https://python-rq.org/docs

Enqueues a job using the command line interface.

```APIDOC
## rq enqueue

### Description
Enqueues a job to a specified queue via the command line. Supports passing functions and arguments with various parsing methods.

### Usage
`rq enqueue [OPTIONS] FUNCTION [ARGUMENTS]`

### Options
- **-q, --queue** (value) - The name of the queue.
- **--timeout** (value) - Maximum runtime of the job before interruption.
- **--result-ttl** (value) - Duration to keep successful jobs and results.
- **--ttl** (value) - Maximum queued time before discarding.
- **--failure-ttl** (value) - Duration to keep failed jobs.
- **--description** (value) - Additional job description.
- **--depends-on** (value) - Job ID that must complete before this job is queued.
- **--job-id** (value) - Unique ID for the job.
- **--at-front** (flag) - Places the job at the front of the queue.
- **--retry-max** (value) - Maximum number of retries.
- **--retry-interval** (value) - Interval between retries in seconds.
- **--schedule-in** (value) - Delay until the function is enqueued (e.g., 10s, 5m, 2d).
- **--schedule-at** (value) - ISO 8601 formatted time to enqueue the job.
- **--quiet** (flag) - Only logs errors.
```

--------------------------------

### Execute jobs synchronously for testing

Source: https://python-rq.org/docs/testing

Use is_async=False to run jobs immediately in the current thread and utilize FakeStrictRedis to avoid needing a live Redis server.

```python
from fakeredis import FakeStrictRedis
from rq import Queue

queue = Queue(is_async=False, connection=FakeStrictRedis())
job = queue.enqueue(my_long_running_job)
assert job.is_finished
```

--------------------------------

### Implement SpawnWorker

Source: https://python-rq.org/docs/workers

Uses os.spawn() instead of os.fork() for job execution, suitable for Windows or newer MacOS environments.

```python
from rq import SpawnWorker, Queue

queue = Queue('default')
worker = SpawnWorker([queue])
worker.work()
```

```bash
rq worker -w rq.worker.SpawnWorker
```

--------------------------------

### Run RQ Cron via CLI

Source: https://python-rq.org/docs/cron

Execute the cron scheduler using a configuration file or module path.

```bash
$ rq cron my_cron_config.py
```

```bash
$ rq cron myapp.cron_config
```

```bash
$ rq cron myapp.cron_config --url redis://localhost:6379/1 --path src
```

--------------------------------

### rq cron CLI

Source: https://python-rq.org/docs/cron

The rq cron command-line interface allows running a cron scheduler from a configuration file or module.

```APIDOC
## CLI Command: rq cron

### Description
Starts the RQ cron scheduler using a specified configuration file or module path.

### Usage
`rq cron [config_path] [options]`

### Options
- **--url, -u** (string) - Optional - Redis connection URL (env: RQ_REDIS_URL)
- **--config, -c** (string) - Optional - Python module with RQ settings (env: RQ_CONFIG)
- **--path, -P** (string) - Optional - Additional Python import paths (can be specified multiple times)
- **--logging-level, -l** (string) - Optional - Logging level (DEBUG, INFO, WARNING, ERROR, CRITICAL; default: INFO)
- **--worker-class, -w** (string) - Optional - Dotted path to RQ Worker class
- **--job-class, -j** (string) - Optional - Dotted path to RQ Job class
- **--queue-class** (string) - Optional - Dotted path to RQ Queue class
- **--connection-class** (string) - Optional - Dotted path to Redis client class
- **--serializer, -S** (string) - Optional - Dotted path to serializer class

### Positional Argument
- **config_path** (string) - Required - File path or module path to your cron configuration
```

--------------------------------

### Configure Redis Sentinel

Source: https://python-rq.org/docs/connections

Define a SENTINEL dictionary in the configuration file to enable fault-tolerant connections to Redis.

```python
SENTINEL: {
    'INSTANCES':[('remote.host1.org', 26379), ('remote.host2.org', 26379), ('remote.host3.org', 26379)],
    'MASTER_NAME': 'master',
    'DB': 2,
    'USERNAME': 'redis-user',
    'PASSWORD': 'redis-secret',
    'SOCKET_TIMEOUT': None,
    'CONNECTION_KWARGS': {  # Eventual addition Redis connection arguments
        'ssl_ca_path': None,
    },
    'SENTINEL_KWARGS': {    # Eventual Sentinels connections arguments
        'username': 'sentinel-user',
        'password': 'sentinel-secret',
    },
}
```

--------------------------------

### queue.enqueue(func, on_success=..., on_failure=..., on_stopped=...)

Source: https://python-rq.org/docs

Enqueues a job with optional callback functions to be executed when the job reaches specific states.

```APIDOC
## queue.enqueue(func, on_success=None, on_failure=None, on_stopped=None)

### Description
Enqueues a function to be executed by a worker, with optional callbacks for success, failure, or stopped states.

### Parameters
- **func** (callable/str) - Required - The function to execute.
- **on_success** (callable/Callback) - Optional - Function called when the job completes successfully.
- **on_failure** (callable/Callback) - Optional - Function called when the job fails.
- **on_stopped** (callable/Callback) - Optional - Function called when the job is stopped.
```

--------------------------------

### Accessing job registries via Queue objects

Source: https://python-rq.org/docs/job_registries

Shows the shorthand properties available on Queue objects to access various job registries directly.

```python
from redis import Redis
from rq import Queue

redis = Redis()
queue = Queue(connection=redis)

queue.started_job_registry  # Returns StartedJobRegistry
queue.deferred_job_registry   # Returns DeferredJobRegistry
queue.finished_job_registry  # Returns FinishedJobRegistry
queue.failed_job_registry  # Returns FailedJobRegistry
queue.scheduled_job_registry  # Returns ScheduledJobRegistry
```

--------------------------------

### Queue Management Methods

Source: https://python-rq.org/docs

Methods for interacting with RQ Queue instances to manage job lifecycle and queue state.

```APIDOC
## Queue Methods

### len(q)
Returns the number of jobs currently in the queue (excluding deferred jobs).

### q.job_ids
Returns a list of job IDs currently in the queue.

### q.jobs
Returns a list of enqueued job instances.

### q.fetch_job(job_id)
Returns the job instance associated with the provided `job_id`.

### q.empty()
Removes all jobs from the queue.

### q.delete(delete_jobs=True)
Deletes the queue. If `delete_jobs` is set to `True`, all jobs within the queue are also removed.
```

--------------------------------

### Job.create(func, *args, **kwargs)

Source: https://python-rq.org/docs/jobs

Creates a job instance without immediately enqueuing it. Requires a subsequent call to enqueue_job().

```APIDOC
## Job.create(func, *args, **kwargs)

### Description
Instantiates a job object. Supported keyword arguments include timeout, result_ttl, ttl, failure_ttl, depends_on, id, description, connection, status, origin, meta, args, and kwargs.

### Example
```python
from rq.job import Job
job = Job.create(func, 'arg1', id='my_job_id')
q.enqueue_job(job)
```
```

--------------------------------

### Run tasks with SimpleWorker

Source: https://python-rq.org/docs/testing

Use SimpleWorker to execute tasks in the same process, avoiding fork() issues with frameworks like Django.

```python
from redis import Redis
from rq import SimpleWorker, Queue

queue = Queue(connection=Redis())
queue.enqueue(my_long_running_job)
worker = SimpleWorker([queue], connection=queue.connection)
worker.work(burst=True)  # Runs enqueued job
# Check for result...
```

--------------------------------

### Bulk Job Enqueueing with RQ

Source: https://python-rq.org/docs

Enqueue multiple jobs at once using enqueue_many and prepare_data. Optionally pass a Redis pipeline to execute operations atomically.

```python
jobs = q.enqueue_many(
  [
    Queue.prepare_data(count_words_at_url, ('http://nvie.com',), job_id='my_job_id'),
    Queue.prepare_data(count_words_at_url, ('http://nvie.com',), job_id='my_other_job_id'),
  ]
)
```

```python
with q.connection.pipeline() as pipe:
  jobs = q.enqueue_many(
    [
      Queue.prepare_data(count_words_at_url, ('http://nvie.com',), job_id='my_job_id'),
      Queue.prepare_data(count_words_at_url, ('http://nvie.com',), job_id='my_other_job_id'),
    ],
    pipeline=pipe
  )
  pipe.execute()
```

--------------------------------

### Queue.enqueue(f, *args, **kwargs)

Source: https://python-rq.org/docs/jobs

Enqueues a function for background execution. Returns a Job instance.

```APIDOC
## Queue.enqueue(f, *args, **kwargs)

### Description
Enqueues a function to be executed by a worker. Returns a Job object representing the enqueued task.

### Parameters
- **f** (callable) - Required - The function to be executed.
- **args** (tuple) - Optional - Positional arguments for the function.
- **kwargs** (dict) - Optional - Keyword arguments for the function.
- **job_id** (str) - Optional - A custom identifier for the job (letters, numbers, underscores, dashes).

### Example
```python
from rq import Queue
from redis import Redis

q = Queue(connection=Redis())
job = q.enqueue(func, 'arg1', job_id='my_job_id')
```
```

--------------------------------

### Register cron-based jobs

Source: https://python-rq.org/docs/cron

Use the cron parameter with standard cron syntax to schedule tasks. Requires version 2.5 or later.

```python
# Every day at 2:30 AM
cron.register(daily_backup, queue_name='maintenance', cron='30 2 * * *')

# Every 15 minutes
cron.register(frequent_task, queue_name='default', cron='*/15 * * * *')

# Weekly on Sundays at 6:00 PM
cron.register(weekly_report, queue_name='reports', cron='0 18 * * 0')
```

--------------------------------

### Display RQ Status with rq info

Source: https://python-rq.org/docs/monitoring

View the current state of queues and active workers using the command line.

```bash
$ rq info
high       |██████████████████████████ 20
low        |██████████████ 12
default    |█████████ 8
3 queues, 45 jobs total

Bricktop.19233 idle: low
Bricktop.19232 idle: high, default, low
Bricktop.18349 idle: default
3 workers, 3 queues
```

--------------------------------

### Access Return Value Shortcut

Source: https://python-rq.org/docs/results

Use the return_value() method as a shortcut to retrieve the return value of the latest successful execution.

```python
job = Job.fetch(id='my_id', connection=redis)
print(job.return_value())  # Shortcut for job.latest_result().return_value
```

--------------------------------

### rq.Callback(func, timeout=60)

Source: https://python-rq.org/docs

Configures a callback function with a specific execution timeout.

```APIDOC
## rq.Callback(func, timeout=60)

### Description
Wraps a callback function to define a custom execution timeout.

### Parameters
- **func** (callable/str) - Required - The function to execute as a callback.
- **timeout** (int/str) - Optional - Execution time limit (default 60 seconds).
```

--------------------------------

### Enqueue a job using a string reference

Source: https://python-rq.org/docs

Use a string path to reference a function when the web process does not have direct access to the worker's source code.

```python
q = Queue('low', connection=redis_conn)
q.enqueue('my_package.my_module.my_func', 3, 4)
```

--------------------------------

### Conditional Job Registration

Source: https://python-rq.org/docs/cron

Leverage standard Python logic within configuration files to conditionally register jobs based on environment variables.

```python
import os
from rq.cron import register
from myapp.tasks import cleanup

if os.getenv("ENABLE_CLEANUP") == "true":
    register(cleanup, queue_name="maintenance", interval=600)
```

--------------------------------

### Group.create(connection=None)

Source: https://python-rq.org/docs

Creates a new group to track multiple jobs under a single ID.

```APIDOC
## Group.create(connection=None)

### Description
Initializes a new job group in Redis.

### Parameters
- **connection** (Redis) - Optional - The Redis connection instance.
```

--------------------------------

### Pass arguments to job functions

Source: https://python-rq.org/docs/jobs

Use the args and kwargs parameters in Job.create to pass data to the underlying function without conflicting with RQ's own configuration parameters.

```python
job = Job.create(count_words_at_url,
          ttl=30,  # This ttl will be used by RQ
          args=('http://nvie.com',),
          kwargs={
              'description': 'Function description', # This is passed on to count_words_at_url
              'ttl': 15  # This is passed on to count_words_at_url function
          })
```

--------------------------------

### Requeue Failed Jobs via CLI

Source: https://python-rq.org/docs/jobs

Uses the RQ CLI tool to requeue specific jobs or all jobs within a queue's failed registry.

```bash
# This will requeue foo_job_id and bar_job_id from myqueue's failed job registry
rq requeue --queue myqueue -u redis://localhost:6379 foo_job_id bar_job_id

# This command will requeue all jobs in myqueue's failed job registry
rq requeue --queue myqueue -u redis://localhost:6379 --all
```

--------------------------------

### Register interval-based jobs

Source: https://python-rq.org/docs/cron

Use the interval parameter to define execution frequency in seconds.

```python
cron.register(my_function, queue_name='default', interval=300)  # Every 5 minutes
cron.register(hourly_task, queue_name='hourly', interval=3600)  # Every hour
```

--------------------------------

### Defining Cron Jobs in Configuration Files

Source: https://python-rq.org/docs/cron

Use the register function within a Python module to define scheduled tasks with intervals, arguments, and timeouts.

```python
# my_cron_config.py
from rq.cron import register
from myapp.tasks import cleanup_database, generate_reports, backup_files

# Simple job - runs every 60 seconds
register(cleanup_database, queue_name='maintenance', interval=60)

# Job with arguments
register(
    generate_reports,
    queue_name='reports',
    args=('daily',),
    kwargs={'format': 'pdf', 'email': True},
    interval=3600
)

# Job with custom timeout and TTL settings
register(
    backup_files,
    queue_name='backup',
    cron='0 2 * * *',  # Daily at 2 AM
    job_timeout=1800,    # 30 minutes max execution time
    result_ttl=3600, # Keep results for 1 hour
    failure_ttl=86400 # Keep failed jobs for 1 day
)
```

--------------------------------

### Configure Queue and Worker with JSON Serializer

Source: https://python-rq.org/docs/workers

Ensure consistency by using the same serializer for both the Queue and the Worker.

```python
from rq import Queue, Worker
from rq.serializers import JSONSerializer

queue = Queue('foo', serializer='json')        # or: serializer=JSONSerializer
worker = Worker([queue], serializer='json')    # or: serializer=JSONSerializer
```

--------------------------------

### Implement custom job and queue classes

Source: https://python-rq.org/docs/workers

Define custom classes and ensure they are used during the enqueueing process to maintain consistency with the worker configuration.

```python
from rq import Queue
from rq.job import Job

class CustomJob(Job):
    pass

class CustomQueue(Queue):
    job_class = CustomJob

queue = CustomQueue('default', connection=redis_conn)
queue.enqueue(some_func)
```

--------------------------------

### Configure dependency failure handling

Source: https://python-rq.org/docs

Use the Dependency class to allow dependent jobs to run even if the parent job fails, and optionally move them to the front of the queue.

```python
from redis import Redis
from rq.job import Dependency
from rq import Queue

queue = Queue(connection=Redis())
job_1 = queue.enqueue(div_by_zero)
dependency = Dependency(
    jobs=[job_1],
    allow_failure=True,    # allow_failure defaults to False
    enqueue_at_front=True  # enqueue_at_front defaults to False
)
job_2 = queue.enqueue(say_hello, depends_on=dependency)

"""
  job_2 will execute even though its dependency (job_1) fails,
  and it will be enqueued at the front of the queue.
"""
```

--------------------------------

### Set Polling Interval for rq info

Source: https://python-rq.org/docs/monitoring

Configure the refresh rate for the console output using the --interval flag.

```bash
$ rq info --interval 1
```

```bash
$ rq info --interval 0.5
```

--------------------------------

### Define Success Callback

Source: https://python-rq.org/docs

Success callbacks must accept job, connection, and result arguments, along with *args and **kwargs for future compatibility.

```python
def report_success(job, connection, result, *args, **kwargs):
    pass
```

--------------------------------

### Verify scheduled job registry

Source: https://python-rq.org/docs/scheduling

Demonstrates that scheduled jobs reside in the ScheduledJobRegistry rather than the standard queue.

```python
from datetime import timedelta
from redis import Redis

from rq import Queue
from rq.registry import ScheduledJobRegistry

queue = Queue(name='default', connection=Redis())
job = queue.enqueue_in(timedelta(seconds=10), say_nothing)
print(job in queue)  # Outputs False as job is not enqueued

registry = ScheduledJobRegistry(queue=queue)
print(job in registry)  # Outputs True as job is placed in ScheduledJobRegistry
```

--------------------------------

### Organize Workers by Queue

Source: https://python-rq.org/docs/monitoring

Toggle between default worker-centric view and queue-centric view using the -R flag.

```bash
$ rq info
...

Mickey.26421 idle: high, default
Bricktop.25458 busy: high, default, low
Turkish.25812 busy: high, default
3 workers, 3 queues
```

```bash
$ rq info -R
...

high:    Bricktop.25458 (busy), Mickey.26421 (idle), Turkish.25812 (busy)
low:     Bricktop.25458 (busy)
default: Bricktop.25458 (busy), Mickey.26421 (idle), Turkish.25812 (busy)
failed:  –
3 workers, 4 queues
```

--------------------------------

### Implementing Concurrency Rate Limits

Source: https://python-rq.org/docs

Use RateLimit to restrict the number of concurrent jobs sharing a specific key. Jobs exceeding the limit are queued until a slot becomes available.

```python
from redis import Redis
from rq import Queue, RateLimit

queue = Queue(connection=Redis())
rate_limit = RateLimit(key='reports', concurrency=2)

job_1 = queue.enqueue(generate_report, 1, rate_limit=rate_limit)
job_2 = queue.enqueue(generate_report, 2, rate_limit=rate_limit)
job_3 = queue.enqueue(generate_report, 3, rate_limit=rate_limit)
```

--------------------------------

### Enqueue Job with Callbacks

Source: https://python-rq.org/docs

Register callback functions directly when enqueuing a job to handle success, failure, or stopped states.

```python
queue.enqueue(say_hello, on_success=report_success, on_failure=report_failure, on_stopped=report_stopped)
```

--------------------------------

### Enqueue Jobs with JSON Serializer

Source: https://python-rq.org/docs/jobs

Configure a queue to use the JSONSerializer to avoid the security risks of the default pickle serializer.

```python
from rq import Queue
from rq.serializers import JSONSerializer

queue = Queue(connection=connection, serializer='json')  # or: serializer=JSONSerializer
job = queue.enqueue('my_module.count_words', 'https://example.com')
```

--------------------------------

### Removing jobs from a registry

Source: https://python-rq.org/docs/job_registries

Illustrates how to iterate through registry job IDs to remove them, with an optional flag to delete the job data entirely.

```python
from redis import Redis
from rq import Queue
from rq.registry import FailedJobRegistry

redis = Redis()
queue = Queue(connection=redis)
registry = FailedJobRegistry(queue=queue)

# This is how to remove a job from a registry
for job_id in registry.get_job_ids():
    registry.remove(job_id)

# If you want to remove a job from a registry AND delete the job,
# use `delete_job=True`
for job_id in registry.get_job_ids():
    registry.remove(job_id, delete_job=True)
```

--------------------------------

### Specify custom worker class via CLI

Source: https://python-rq.org/docs/workers

Use the -w option to point to a custom worker class implementation.

```bash
$ rq worker -w 'path.to.GeventWorker'
```

--------------------------------

### Implement SimpleWorker

Source: https://python-rq.org/docs/workers

Executes jobs in the same process without forking. Not recommended for production due to lack of heartbeats.

```python
from rq import SimpleWorker, Queue

queue = Queue('default')
worker = SimpleWorker([queue])
worker.work()
```

```bash
rq worker -w rq.worker.SimpleWorker
```

--------------------------------

### rq worker-pool

Source: https://python-rq.org/docs/workers

CLI command to run multiple workers in a single process.

```APIDOC
## rq worker-pool

### Description
Allows you to run multiple workers in a single CLI command.

### Usage
`rq worker-pool [queues] [options]`

### Options
- **-u, --url** (str) - Redis connection URL.
- **-w, --worker-class** (str) - Path to the Worker class.
- **-n, --num-workers** (int) - Number of workers to spawn (defaults to 2).
- **-b, --burst** (flag) - Run workers in burst mode.
- **-l, --logging-level** (str) - Logging level (DEBUG, INFO, WARNING, ERROR, CRITICAL).
- **-S, --serializer** (str) - Serializer type (json, pickle, or import path).
- **-P, --path** (str) - Import path for tasks.
- **-j, --job-class** (str) - Path to the Job class.
- **--exception-handler** (str) - Path to exception handler class.
- **-c, --config** (str) - Module containing RQ settings.
```

--------------------------------

### Configure custom exception handlers

Source: https://python-rq.org/docs/workers

Specify one or more custom exception handler paths to override or extend default error handling behavior.

```bash
$ rq worker --exception-handler 'path.to.my.ErrorHandler'

# Multiple exception handlers is also supported
$ rq worker --exception-handler 'path.to.my.ErrorHandler' --exception-handler 'another.ErrorHandler'
```

--------------------------------

### Programmatic Cron Scheduling

Source: https://python-rq.org/docs/cron

Initialize a CronScheduler and register tasks using the Python API.

```python
from redis import Redis
from rq.cron import CronScheduler
from myapp.tasks import cleanup_database, send_reports

# Create a cron scheduler
redis_conn = Redis(host='localhost', port=6379, db=0)
cron = CronScheduler(connection=redis_conn, logging_level='INFO')

# Register jobs
cron.register(
    cleanup_database,
    queue_name='maintenance',
    interval=3600
)

cron.register(
    send_reports,
    queue_name='reports',
    args=('daily',),
    cron='0 9 * * *'  # Daily at 9 AM
)

# Start the scheduler (this will block until interrupted)
try:
    cron.start()
except KeyboardInterrupt:
    print("Shutting down cron scheduler...")
```

--------------------------------

### RateLimit(key, concurrency)

Source: https://python-rq.org/docs

Defines a concurrency limit for jobs sharing a specific key.

```APIDOC
## RateLimit(key, concurrency)

### Description
Creates a rate limit object to cap the number of concurrent jobs sharing the same key.

### Parameters
- **key** (str) - Required - The identifier for the shared limit.
- **concurrency** (int) - Required - The maximum number of concurrent jobs allowed.
```

--------------------------------

### Queue with Custom Serializer

Source: https://python-rq.org/docs/jobs

Configuring a queue to use a specific serializer for job arguments.

```APIDOC
## Queue(connection, serializer)

### Description
Initializes a queue with a custom serializer. The serializer must implement `loads` and `dumps`.

### Parameters
- **connection** (Redis) - Required - Redis connection instance.
- **serializer** (str or class) - Required - The serializer to use (e.g., 'json' or JSONSerializer).
```

--------------------------------

### Use Connection context manager

Source: https://python-rq.org/docs/connections

The Connection context manager allows setting a default Redis connection for objects created within its scope. Note that this is deprecated in favor of explicit connection management.

```python
from rq import Queue, Connection
from redis import Redis

with Connection(Redis('localhost', 6379)):
    q1 = Queue('foo')
    with Connection(Redis('remote.host.org', 9836)):
        q2 = Queue('bar')
    q3 = Queue('qux')

assert q1.connection != q2.connection
assert q2.connection != q3.connection
assert q1.connection == q3.connection
```

--------------------------------

### Manage worker activity via CLI

Source: https://python-rq.org/docs

Use the rq suspend and resume commands to control whether workers pick up new jobs. Suspending does not interrupt currently running jobs.

```bash
# Suspend indefinitely
rq suspend

# Suspend for a specific duration (in seconds) then automatically
# resume work again.
rq suspend --duration 300

# Resume work again.
rq resume
```

--------------------------------

### Configuring job TTL settings

Source: https://python-rq.org/docs/jobs

Define result_ttl and ttl during job creation or enqueueing to control how long results are kept and how long a job can remain queued.

```python
# When creating the job:
job = Job.create(func=say_hello,
                 result_ttl=600,  # how long (in seconds) to keep the job (if successful) and its results
                 ttl=43,  # maximum queued time (in seconds) of the job before it's discarded.
                )

# or when queueing a new job:
job = q.enqueue(count_words_at_url,
                'http://nvie.com',
                result_ttl=600,  # how long to keep the job (if successful) and its results
                ttl=43  # maximum queued time
               )
```

--------------------------------

### Inspect Job Results

Source: https://python-rq.org/docs/results

Fetch a job and inspect its latest result type and associated data.

```python
job = Job.fetch(id='my_id', connection=redis)
result = job.latest_result()  #  returns Result(id=uid, type=SUCCESSFUL)
if result.type == result.Type.SUCCESSFUL:
    print(result.return_value)
elif result.type == result.Type.FAILED:
    print(result.exc_string)
```

--------------------------------

### Query Specific Queues

Source: https://python-rq.org/docs/monitoring

Filter the rq info output to display statistics for specific queues only.

```bash
$ rq info high default
high       |██████████████████████████ 20
default    |█████████ 8
2 queues, 28 jobs total

Bricktop.19232 idle: high, default
Bricktop.18349 idle: default
2 workers, 2 queues
```

--------------------------------

### Configure Webhooks for Job Enqueueing

Source: https://python-rq.org/docs

Pass a list of Webhook objects to the webhooks argument when enqueuing a job to trigger notifications on specific terminal states.

```python
from rq import Webhook

queue.enqueue(say_hello,
              webhooks=[Webhook('https://example.com/finished', job_status='finished'),
                        Webhook('https://example.com/failed', job_status='failed', method='POST')])
```

--------------------------------

### Handling Variable Intervals

Source: https://python-rq.org/docs/scheduling

Illustrates how the last interval in a list is reused if the number of repetitions exceeds the list length.

```python
# This job will repeat 5 times with these intervals:
# 1st repetition: after 5 seconds
# 2nd repetition: after 10 seconds
# 3rd repetition: after 15 seconds
# 4th repetition: after 15 seconds
# 5th repetition: after 15 seconds
job = queue.enqueue(my_function, repeat=Repeat(times=5, interval=[5, 10, 15]))
```

--------------------------------

### Job.fetch(job_id, connection=None)

Source: https://python-rq.org/docs/jobs

Retrieves an existing job from Redis by its ID.

```APIDOC
## Job.fetch(job_id, connection=None)

### Description
Fetches a job instance from Redis to inspect its attributes or status.

### Parameters
- **job_id** (str) - Required - The ID of the job to retrieve.
- **connection** (Redis) - Optional - The Redis connection instance.

### Example
```python
from rq.job import Job
job = Job.fetch('my_job_id', connection=redis_conn)
print(job.get_status())
```
```

--------------------------------

### Schedule job after a time delay

Source: https://python-rq.org/docs/scheduling

Uses enqueue_in to schedule a job to run after a specified timedelta.

```python
from datetime import timedelta
from rq import Queue
from redis import Redis
from somewhere import say_hello

queue = Queue(name='default', connection=Redis())

# Schedules job to be run in 10 seconds
job = queue.enqueue_in(timedelta(seconds=10), say_hello)
```

--------------------------------

### Storing custom metadata on a job

Source: https://python-rq.org/docs/jobs

Use the meta dictionary to store arbitrary pickleable data, then call save_meta to persist changes to the job.

```python
import socket

def add(x, y):
    job = get_current_job()
    job.meta['handled_by'] = socket.gethostname()
    job.save_meta()

    # do more work
    time.sleep(1)
    return x + y
```

--------------------------------

### Configure Worker with JSON Serializer

Source: https://python-rq.org/docs/workers

Pass the 'json' shorthand or the JSONSerializer class to the Worker constructor to handle JSON-serialized jobs.

```python
from rq import Worker
from rq.serializers import JSONSerializer

worker = Worker('foo', serializer='json')  # or: serializer=JSONSerializer
```

--------------------------------

### Fetch a job by ID

Source: https://python-rq.org/docs/jobs

Retrieve job details from Redis using the Job.fetch method.

```python
from redis import Redis
from rq.job import Job

redis = Redis()
job = Job.fetch('my_job_id', connection=redis)
print('Status: %s' % job.get_status())
```

--------------------------------

### Registering Cron Jobs with Webhooks

Source: https://python-rq.org/docs/cron

Use the Webhook class to notify external endpoints upon job completion or failure.

```python
from rq import Webhook

cron.register(
    backup_database,
    queue_name='maintenance',
    cron='0 2 * * *',
    webhooks=[
        Webhook('https://example.com/finished', 'finished'),
        Webhook('https://example.com/failed', 'failed'),
    ],
)
```

--------------------------------

### Implement RoundRobinWorker

Source: https://python-rq.org/docs/workers

Dequeues jobs from multiple queues in a round-robin fashion for fair scheduling.

```python
from rq import RoundRobinWorker, Queue

queues = [Queue('q1'), Queue('q2'), Queue('q3')]
worker = RoundRobinWorker(queues)
worker.work()
```

```bash
rq worker -w rq.worker.RoundRobinWorker q1 q2 q3
```

--------------------------------

### Configure Callback Timeouts

Source: https://python-rq.org/docs

Use the Callback object to specify custom execution timeouts for individual job callbacks.

```python
from rq import Callback
queue.enqueue(say_hello,
              on_success=Callback(report_success),  # default callback timeout (60 seconds)
              on_failure=Callback(report_failure, timeout=10), # 10 seconds timeout
              on_stopped=Callback(report_stopped, timeout="2m")) # 2 minute timeout
```

--------------------------------

### Unsupported Redis encoding

Source: https://python-rq.org/docs/connections

The decode_responses=True argument is not supported by RQ due to internal encoding/decoding mechanisms.

```python
from redis import Redis
from rq import Queue

conn = Redis(..., decode_responses=True) # This is not supported
q = Queue(connection=conn)
```

--------------------------------

### queue.enqueue_many(job_datas, pipeline=None)

Source: https://python-rq.org/docs

Enqueues multiple jobs in bulk. Optionally accepts a Redis pipeline to execute operations atomically.

```APIDOC
## queue.enqueue_many(job_datas, pipeline=None)

### Description
Enqueues multiple jobs in bulk. This is useful for reducing the number of round-trips to Redis.

### Parameters
- **job_datas** (list) - Required - A list of job data objects prepared via `Queue.prepare_data`.
- **pipeline** (redis.client.Pipeline) - Optional - A Redis pipeline object to group the enqueue operations.
```

--------------------------------

### Accessing Failed Jobs with FailedJobRegistry

Source: https://python-rq.org/docs/exceptions

Use the FailedJobRegistry to retrieve and inspect exception information for jobs that failed during execution.

```python
from redis import Redis
from rq import Queue
from rq.job import Job
from rq.registry import FailedJobRegistry

redis = Redis()
queue = Queue(connection=redis)
registry = FailedJobRegistry(queue=queue)

# Show all failed job IDs and the exceptions they caused during runtime
for job_id in registry.get_job_ids():
    job = Job.fetch(job_id, connection=redis)
    print(job_id, job.exc_info)
```

--------------------------------

### Job.fetch_many(job_ids, connection=None)

Source: https://python-rq.org/docs/jobs

Efficiently retrieves multiple jobs from Redis by their IDs.

```APIDOC
## Job.fetch_many(job_ids, connection=None)

### Description
Fetches a list of job instances from Redis based on a list of provided IDs.

### Parameters
- **job_ids** (list) - Required - A list of job ID strings.
- **connection** (Redis) - Optional - The Redis connection instance.

### Example
```python
jobs = Job.fetch_many(['id1', 'id2'], connection=redis_conn)
```
```

--------------------------------

### Run Worker Pool

Source: https://python-rq.org/docs/workers

Executes multiple workers in a single CLI command.

```bash
rq worker-pool high default low -n 3
```

--------------------------------

### Enqueue a job to a specific queue

Source: https://python-rq.org/docs

Specify a queue name to distribute work across different priority levels or categories.

```python
q = Queue('low', connection=redis_conn)
q.enqueue(count_words_at_url, 'http://nvie.com')
```
Tokens:
Raw