Flask-RQ

When you decorate a function with @RQ.job, it returns a JobWrapper instance. While you should not import JobWrapper directly, you can use its public methods to schedule the decorated function as a background task.

Key methods available on the decorated function include:

• enqueue(): Enqueue the function for immediate execution.
• enqueue_at(datetime): Enqueue the function to run at a specific time.
• enqueue_in(timedelta): Enqueue the function to run after a specific delay.
• cron_register(...): Register the function as a recurring task using cron syntax.
• __call__(): Calling the function directly executes it synchronously (not as a background job).

from flask_rq import RQ

rq = RQ()

@rq.job
def my_background_task(arg1, arg2):
    print(f"Running task with {arg1} and {arg2}")

# Usage examples:
# 1. Immediate execution
my_background_task.enqueue(1, 2)

# 2. Scheduled execution
from datetime import datetime, timedelta
my_background_task.enqueue_at(datetime(2025, 1, 1, 12, 0, 0), 1, 2)

# 3. Delayed execution
my_background_task.enqueue_in(timedelta(minutes=5), 1, 2)

The @rq.job(queue="...") decorator wraps a function to provide convenient .enqueue() methods that automatically use the extension instance and the specified queue name.

Benefits:

• Type Safety: The wrapper retains the static type signature of the original function, allowing type checkers to validate arguments (unlike the generic queue.enqueue call).
• Direct Execution: You can call the wrapped function directly (e.g., await func(...)) to execute it synchronously/locally instead of enqueuing it.

Warning: Because of how RQ references jobs, you cannot pass a decorated function directly into queue.enqueue. Instead, use the wrapper's own .enqueue() method or pass the underlying function via the .func attribute.

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

# Use the wrapper's method (Recommended)
send_password_reset.enqueue(user_id=user.id)

# Or use the underlying function with a queue
# rq.queues["email"].enqueue(send_password_reset.func, user_id=user.id)

# To execute the function directly (not as a job)
await send_password_reset(user_id=user.id)

For high-fidelity testing that mimics production, run a real local RQ server. The recommended approach using pytest is to:

1. Use a session scoped fixture to start redis-server in a subprocess using a reserved ephemeral port.
2. Use a function scoped fixture to call .flushall() on the Redis instance after each test to ensure a clean state.

This requires the ephemeral-port-reserve package to avoid port conflicts.

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()