Flask-RQ

https://flask-rq.readthedocs.io
Admin
Flask-RQ is a Flask/Quart extension for background job execution using RQ, enabling long-running...
Tokens:6,811
Snippets:67
Trust Score:8.3
Update:10 months ago
Show doc for...
Tokens:
Raw
### Configure GitHub Actions for Redis Setup

Source: https://flask-rq.readthedocs.io/en/stable/testing

This configuration snippet demonstrates how to set up Redis for use in GitHub Actions. It utilizes the 'shogo82148/actions-setup-redis' action to install Redis. Crucially, 'auto-start: false' is specified because the tests themselves are responsible for starting the Redis server.

```yaml
steps:
  # ...
  - uses: shogo82148/actions-setup-redis@v1
    with:
      auto-start: false
  # ...

```

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

### Initialize Flask-RQ Instance

Source: https://flask-rq.readthedocs.io/en/stable/start

Demonstrates how to initialize the Flask-RQ extension with a Flask or Quart application. The `RQ` instance can be initialized immediately with the application object or later using the `init_app()` method, which is useful for application factories. This setup allows for configuring queues and connections to Redis.

```python
from flask_rq import RQ

rq = RQ()
rq.init_app(app)  # call in the app factory if you're using that pattern

```

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

### Run Redis Servers Locally using Docker

Source: https://flask-rq.readthedocs.io/en/stable/server

Demonstrates how to start Redis, Valkey, or Redict servers locally using Docker. These commands pull the latest images and map the default Redis port (6379) to the host machine.

```bash
# Redis
$ docker run -d --rm -p 6379:6379 redis:latest

# Valkey
$ docker run -d --rm -p 6379:6379 valkey/valkey:latest

# Redict
$ docker run -d --rm -p 6379:6379 registry.redict.io/redict:latest

```

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

### Run Flask-RQ Worker

Source: https://flask-rq.readthedocs.io/en/stable/start

Provides the command-line instruction to start a Flask-RQ worker. This worker is essential for processing jobs enqueued by the application. Ensure a Redis server is running and accessible.

```bash
$ flask rq worker

```

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

### Starting a Flask-RQ Worker with Scheduler Enabled

Source: https://flask-rq.readthedocs.io/en/stable/job

Demonstrates the command-line instruction to start a Flask-RQ worker that includes the scheduler. This is necessary for processing scheduled jobs that are set to run at specific times or after certain intervals.

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

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

### Start Flask-RQ Worker via CLI

Source: https://flask-rq.readthedocs.io/en/stable/worker

The `flask rq worker` command is the simplest way to start a worker process. This command ensures that jobs run within the application context, making Flask extensions and configurations available. It accepts many options similar to RQ's own `rq worker` command; consult `--help` for details.

```bash
flask rq worker

```

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

### Start and Reset Redis Server for Pytest Tests

Source: https://flask-rq.readthedocs.io/en/stable/testing

This fixture manages a Redis server for Pytest test sessions. It starts 'redis-server' in a subprocess on a reserved ephemeral port at the beginning of the session and terminates it afterward. A separate fixture resets the Redis database by flushing all keys after each test function, ensuring a clean state for isolated testing.

```python
import collections.abc as cabc
import subprocess
import time

import ephemeral_port_reserve
import pytest
from redis import Redis
from redis.exceptions import ConnectionError as RedisConnectionError

@pytest.fixture(scope="session")
def redis_port() -> int:
    return ephemeral_port_reserve.reserve()  # type: ignore[no-any-return]


@pytest.fixture(scope="session", autouse=True)
def _start_redis(
    tmp_path_factory: pytest.TempPathFactory, redis_port: int
) -> cabc.Iterator[None]:
    proc = subprocess.Popen(
        ["redis-server", "--port", str(redis_port)],
        cwd=tmp_path_factory.mktemp("redis-server"),
        stdout=subprocess.DEVNULL,
        stderr=subprocess.DEVNULL,
    )

    while True:
        try:
            redis = Redis(port=redis_port, single_connection_client=True)
            redis.ping()
            break
        except (ConnectionError, RedisConnectionError):  # pragma: no cover
            time.sleep(0.1)

    yield
    proc.terminate()
    proc.wait()


@pytest.fixture(autouse=True)
def _reset_redis(redis_port: int) -> cabc.Iterator[None]:
    yield
    Redis(port=redis_port, single_connection_client=True).flushall()

```

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

### Run Redis Server with Persistent Data using Docker

Source: https://flask-rq.readthedocs.io/en/stable/server

Shows how to run a Redis server using Docker with data persistence enabled. This configuration uses a Docker volume for data storage and enables append-only file (AOF) saving for durability.

```bash
$ docker run -d --rm -p 6379:6379 --volume redis:/data \
    redis:latest redis-server --appendonly yes

```

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

### Use `@rq.job` Decorator for Enqueuing

Source: https://flask-rq.readthedocs.io/en/stable/start

Illustrates a convenient shortcut for enqueuing jobs using the `@rq.job()` decorator. Applying this decorator to a function automatically adds an `enqueue` method to it, which uses the Flask-RQ extension and the specified queue. This simplifies job execution by abstracting the `enqueue` call.

```python
@rq.job(queue="email")
def send_password_reset(user_id: int) -> None:
    ... # Your job logic here

send_password_reset.enqueue(user.id)

```

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

### Enqueue RQ Jobs with Flask-RQ

Source: https://flask-rq.readthedocs.io/en/stable/start

Shows how to execute Python callables (sync or async) as RQ jobs using Flask-RQ. Jobs can be enqueued on the default queue or named queues via the `RQ.queue` attribute or `RQ.queues` dictionary. Arguments must be pickleable. Jobs are automatically run within the application's context.

```python
def send_password_reset(user_id: int) -> None:
    ... # Your job logic here

# Enqueue on the default queue
rq.queue.enqueue(send_password_reset, user.id)

# Enqueue on a named queue
rq.queues["email"].enqueue(send_password_reset, user.id)

```

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

### Enqueuing Jobs from Async Functions using asyncio.to_thread

Source: https://flask-rq.readthedocs.io/en/stable/job

Provides an example of how to enqueue a job from an asynchronous context, such as an `async def` view function, when direct calls to `rq.enqueue` might be perceived as blocking. `asyncio.to_thread` is used to run the potentially blocking `enqueue` operation in a separate thread.

```python
await asyncio.to_thread(rq.enqueue, send_password_reset, user_id=user.id)
```

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

### Specify Queues for Flask-RQ Worker CLI

Source: https://flask-rq.readthedocs.io/en/stable/worker

When starting a worker using the `flask rq worker` command, you can specify which queues the worker should monitor. If no queues are specified, the worker monitors all queues defined in `RQ_QUEUES`. By listing queue names, the worker will only watch those specific queues and use the connection of the first listed queue.

```bash
# all queues, default's connection
$ flask rq worker

# two queues, email's connection
$ flask rq worker email email-priority

```

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

### Get or create an RQ queue (Python)

Source: https://flask-rq.readthedocs.io/en/stable/old

The `flask_rq.get_queue` function retrieves an RQ queue by name. If additional keyword arguments are provided, it creates and returns a new queue instance. This function is deprecated and will be removed in Flask-RQ 1.0.

```python
from flask_rq import get_queue

# Get the default queue
default_queue = get_queue()

# Get a specific queue
low_queue = get_queue('low')

# Create a new queue with custom connection arguments
custom_queue = get_queue('custom', host='localhost', port=6380, db=0)
```

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

### Get an RQ worker (Python)

Source: https://flask-rq.readthedocs.io/en/stable/old

The `flask_rq.get_worker` function creates and returns a RQ worker instance that monitors the specified queues. This function is deprecated and will be removed in Flask-RQ 1.0.

```python
from flask_rq import get_worker

# Get a worker for the default and low queues
worker = get_worker('default', 'low')

# Start the worker (typically done externally)
# worker.work()
```

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

### Flask-RQ Configuration

Source: https://flask-rq.readthedocs.io/en/stable/old

Details on how to configure Flask-RQ queues using Flask's configuration system. Configuration keys follow the pattern `RQ_{queue}_{arg}`.

```APIDOC
## Configuration

### Description
Configuration keys use the format `RQ_{queue}_{arg}` to configure Redis connection arguments for specific queues. `RQ_{queue}_URL` can be used for URL-based configuration. The 'default' queue connects to a local Redis server if not otherwise specified. If a queue is not configured, it inherits the default connection settings.

### Example Configuration
```python
app.config |= {
    "RQ_DEFAULT_HOST": "dev.example",
    "RQ_LOW_HOST": "dev.example",
    "RQ_LOW_DB": "1",
    "RQ_HIGH_URL": "redis://dev.example:6379/2",
}
```

### Configuration Keys
- **RQ_{queue}_HOST**: Hostname for the Redis server for the specified queue.
- **RQ_{queue}_DB**: Database number for the Redis server for the specified queue.
- **RQ_{queue}_URL**: Redis connection URL for the specified queue.
- **RQ_{queue}_{arg}**: Any other Redis connection arguments can be configured this way.
```

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

### Queueing Sync and Async Functions with Flask-RQ

Source: https://flask-rq.readthedocs.io/en/stable/job

Demonstrates how to enqueue both synchronous and asynchronous Python functions as jobs using Flask-RQ. Flask-RQ ensures that these jobs run within the application's context, making extensions and databases accessible. The `enqueue` method accepts the function and its arguments.

```python
def update_stats(data):
    ...

async def send_password_reset(user_id):
    ...

rq.queue.enqueue(update_stats, data=...)
rq.queues["email"].enqueue(send_passord_reset, user_id=user.id)
```

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

### Decorate Functions as RQ Jobs

Source: https://flask-rq.readthedocs.io/en/stable/api

Demonstrates how to use the `RQ.job` decorator to wrap functions for background execution via RQ. It shows direct decoration, decoration with arguments for specifying a queue, and using the decorator as a function to wrap lambda expressions.

```python
import flask_rq

@flask_rq.rq.job
def add(a, b):
    return a + b

@flask_rq.rq.job(queue="math")
def sub(a, b):
    return a - b

mul = flask_rq.rq.job(lambda a, b: a * b, queue="math")

```

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

### Create Custom Flask-RQ Worker Script

Source: https://flask-rq.readthedocs.io/en/stable/worker

For advanced customization beyond the CLI options, a Python script can be used to create and run a worker. This script must import the Flask app and the Flask-RQ extension, establish an application context, and then instantiate and run the worker using `RQ.make_worker()`. This method allows for fine-grained control over worker arguments and behavior.

```python
# my_worker.py
from my_app import create_app, rq

app = create_app()

with app.app_context():
    worker = rq.make_worker()
    worker.work()

```

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

### Configure Flask-RQ queues (Python)

Source: https://flask-rq.readthedocs.io/en/stable/old

Flask-RQ allows configuration of Redis queues using Flask's config object with keys prefixed by `RQ_`. Connection arguments or a URL can be specified for each queue. The 'default' queue has a fallback to a local Redis server.

```python
app.config |= {
    "RQ_DEFAULT_HOST": "dev.example",
    "RQ_LOW_HOST": "dev.example",
    "RQ_LOW_DB": "1",
    "RQ_HIGH_URL": "redis://dev.example:6379/2",
}
```

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

### Configure Flask-RQ to Use FakeRedis

Source: https://flask-rq.readthedocs.io/en/stable/testing

This configuration snippet shows how to direct Flask-RQ to use the FakeRedis library for testing. By setting the 'RQ_CONNECTION_CLASS' application configuration variable to 'fakeredis.FakeRedis', tests can run against an in-memory Redis implementation instead of a separate Redis server. Note that some applications might encounter issues with FakeRedis.

```python
app.config["RQ_CONNECTION_CLASS"] = "fakeredis.FakeRedis"

```

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

### Configure Multiple Queues with Flask-RQ

Source: https://flask-rq.readthedocs.io/en/stable/queue

Define additional named queues beyond the default one. This configuration tells Flask-RQ which queues to recognize, in addition to the automatically configured 'default' queue. All listed queues will share the default Redis connection unless otherwise specified.

```python
RQ_QUEUES = ["email", "priority"]
```

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

### Calling Wrapped Functions Directly

Source: https://flask-rq.readthedocs.io/en/stable/job

Shows that a function decorated with `@rq.job` can still be called directly as a regular function, in addition to being enqueued. This allows for immediate execution when needed, separate from background job processing.

```python
await send_password_reset(user_id=user.id)
```

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

### Enqueue Jobs with Flask-RQ

Source: https://flask-rq.readthedocs.io/en/stable/queue

Demonstrates how to add jobs to the default queue or named queues. The `enqueue` method on a queue object takes the function to be executed and its arguments. Jobs are then processed in the background by RQ workers.

```python
rq.queue.enqueue(update_stats, data=...)
rq.queues["email"].enqueue(send_passord_reset, user_id=user.id)
```

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

### Flask-RQ Job Wrapper API

Source: https://flask-rq.readthedocs.io/en/stable/old

Provides functions to wrap functions as RQ jobs, adding enqueue capabilities. Note that these are deprecated since version 0.3 and will be removed in Flask-RQ 1.0.

```APIDOC
## POST /flask_rq.job

### Description
Wraps a decorated function to add an `enqueue` method. `job.enqueue()` is a shortcut for `rq.queue.enqueue(job)`. 

### Method
POST (Implicit, as it's a function wrapper)

### Endpoint
`/flask_rq.job`

### Parameters
#### Path Parameters
None

#### Query Parameters
None

#### Request Body
- **_func_or_queue** (Callable | str | None) - Required/Optional - A job function to wrap, or a queue name to return a new decorator.

### Request Example
```json
{
  "_func_or_queue": "your_job_function" 
}
```

### Response
#### Success Response (200)
- **JobWrapper** - A wrapper object for the job.

#### Response Example
```json
{
  "job_wrapper": "<JobWrapper object>"
}
```
```

```APIDOC
## GET /flask_rq.get_queue

### Description
Retrieves an RQ queue by name. If additional arguments are provided, a new queue instance is created. Uses the existing connection if the named queue is found.

### Method
GET (Implicit, as it's a function call)

### Endpoint
`/flask_rq.get_queue`

### Parameters
#### Path Parameters
None

#### Query Parameters
- **name** (str) - Required - The name of the queue to retrieve. Defaults to 'default'.
- **kwargs** (Any) - Optional - Arguments to create a new queue instance.

### Request Example
```json
{
  "name": "default"
}
```

### Response
#### Success Response (200)
- **Queue** - The RQ queue object.

#### Response Example
```json
{
  "queue": "<rq.Queue object>"
}
```
```

```APIDOC
## GET /flask_rq.get_worker

### Description
Retrieves an RQ worker that monitors the specified queue names.

### Method
GET (Implicit, as it's a function call)

### Endpoint
`/flask_rq.get_worker`

### Parameters
#### Path Parameters
None

#### Query Parameters
- **queues** (str) - Required - Names of the queues for the worker to watch.

### Request Example
```json
{
  "queues": ["default", "high"]
}
```

### Response
#### Success Response (200)
- **Worker** - The RQ worker object.

#### Response Example
```json
{
  "worker": "<rq.Worker object>"
}
```
```

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

### Share Redis Connections Between Flask-RQ Queues

Source: https://flask-rq.readthedocs.io/en/stable/queue

Configure multiple queues to share Redis connections, either the default connection or custom-defined ones. This is useful for managing resources efficiently when certain queues can share the same Redis instance.

```python
RQ_CONNECTION = "redis://redis.my-app.example"
RQ_QUEUES = ["priority", "email", "email-priority"]
RQ_QUEUE_CONNECTIONS = {
    "email": "redis://email-redis.my-app.example",
    "email-priority": "email"
}
```

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

### Configure Separate Redis Connections for Flask-RQ Queues

Source: https://flask-rq.readthedocs.io/en/stable/queue

Define specific Redis connection URLs for individual named queues. This allows different queues to connect to different Redis instances, useful for isolating workloads or using dedicated Redis servers.

```python
RQ_QUEUE_CONNECTIONS = {"email": "redis://email-redis.my-app.example"}
```

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

### Using the @rq.job Decorator for Automatic Enqueuing

Source: https://flask-rq.readthedocs.io/en/stable/job

Illustrates the use of the `@rq.job` decorator to automatically add an `enqueue` method to a function. This decorator simplifies job submission by binding the function to a specific queue. The `enqueue` method preserves the function's type signature for better type checking.

```python
@rq.job(queue="email")
async def send_password_reset(user_id: int) -> None:
    ...

send_password_reset.enqueue(user_id=user.id)

# same as
rq.queues["email"].enqueue(send_password_reset, user_id=user.id)
```

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

### Configure Custom Redis Connection for Flask-RQ

Source: https://flask-rq.readthedocs.io/en/stable/queue

Specify a custom Redis server connection URL for Flask-RQ. By default, Flask-RQ connects to a local Redis instance. This setting allows you to direct all queues (or specific queues via `RQ_QUEUE_CONNECTIONS`) to a different Redis server.

```python
RQ_CONNECTION = "redis://redis.my-app.example"
```

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

### Use @rq.job Decorator to Enqueue Functions in Flask-RQ

Source: https://flask-rq.readthedocs.io/en/stable/queue

The `@rq.job` decorator simplifies enqueuing. It wraps a function, providing it with an `.enqueue()` method that automatically adds the function to the specified queue. This avoids manually calling `rq.queues[...].enqueue(...)`.

```python
@rq.job(queue="email")
def send_password_reset(user_id: int) -> None:
    ...

send_password_reset.enqueue(user_id=user.id)
```

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

### Override Queue in Flask-RQ @rq.job Decorator

Source: https://flask-rq.readthedocs.io/en/stable/queue

Shows how to enqueue a job decorated with `@rq.job` to a different queue than the one specified during decoration. This is achieved by calling `enqueue` on a different queue object from `rq.queues`.

```python
rq.queues["priority"].enqueue(send_password_reset, user_id=user.id)
```

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

### Wrap decorated function with enqueue method (Python)

Source: https://flask-rq.readthedocs.io/en/stable/old

The `flask_rq.job` decorator wraps a function, adding an `enqueue` method for easier job submission to RQ. It can accept a function directly or a queue name to return a new decorator. This API is deprecated and will be removed in Flask-RQ 1.0.

```python
import flask_rq

@flask_rq.job
def my_task(x, y):
    return x + y

# Enqueue the task
job = my_task.enqueue(1, 2)
```
Tokens:
Raw