arq: Async Redis Job Queue

Warnings

• breaking arq underwent a complete rewrite in v0.16, fundamentally changing how workers are registered, and jobs are enqueued and processed. Code written for v0.15 or earlier is incompatible.
  Fix: Users migrating from v0.15 or older must entirely rewrite their arq integration following the new API. Refer to the v0.16 documentation for the new patterns.
• breaking Support for Python 3.8 was officially removed, while support for Python 3.13 was added.
  Fix: Ensure your project runs on Python 3.9 or newer. Upgrade your Python environment if currently using 3.8.
• gotcha All job functions defined for arq workers must accept `ctx` as their first argument, which is a dictionary for passing shared resources (e.g., database connections, HTTP sessions) to jobs.
  Fix: Modify your job function signatures to `async def my_job(ctx, *args, **kwargs):`. Access shared resources via `ctx['key']`.
• gotcha arq uses 'pessimistic execution': jobs are only removed from the queue upon successful completion or final failure. If a worker process shuts down unexpectedly, in-progress jobs will remain in the queue to be rerun by another worker or when the worker restarts.
  Fix: Design your jobs to be idempotent or handle potential duplicate executions gracefully. Be aware that job execution might not be 'exactly once' in failure scenarios.
• gotcha For Python 3.14 and newer, `asyncio.get_event_loop()` no longer implicitly creates an event loop and will raise a `RuntimeError` if no loop is running. While `arq` v0.27.1+ includes internal fixes, custom asyncio setups might be affected.
  Fix: Ensure you are using `arq` v0.27.1 or newer for Python 3.14+ compatibility. If manually managing event loops, use `asyncio.new_event_loop()` explicitly or ensure a loop is set before calling `arq` components.
• gotcha The `ArqRedis` methods `get_all_job_results()` and `queued_jobs()` use inefficient Redis commands (`KEYS` or multiple `GET` operations) internally, making them unsuitable for production environments with a large number of keys or jobs.
  Fix: Avoid using these methods for retrieving large sets of job data in production. If you need to monitor or manage many jobs, consider designing custom Redis queries or using `arq`'s built-in monitoring tools if available for specific use cases.

Install

• pip install arq httpx Install arq and its common dependencies

Imports

• create_pool

  from arq import create_pool

• RedisSettings

  from arq.connections import RedisSettings

• WorkerSettings

  from arq.worker import WorkerSettings

• func

  from arq.worker import func

  Used to wrap job functions for more advanced settings (e.g., `keep_result`, `timeout`).
• cron

  from arq.cron import cron

  Used to define recurring jobs with cron-like schedules.

Quickstart

This quickstart demonstrates how to enqueue jobs using `arq` and how to define `WorkerSettings` for processing them. It includes a `download_content` job, along with `startup` and `shutdown` hooks for resource management (e.g., an `httpx.AsyncClient` session). The script can be run to enqueue jobs, and a separate worker process (using the `arq` CLI) can be started to consume them from Redis. Ensure a Redis server is running and accessible via `ARQ_REDIS_DSN` environment variable or the default `redis://localhost:6379`.

import asyncio
import os

from arq import create_pool
from arq.connections import RedisSettings
from arq.worker import WorkerSettings

# For demonstration, typically use an AsyncClient from httpx or aiohttp
# Install 'httpx' for this example: pip install httpx
import httpx

REDIS_DSN = os.environ.get('ARQ_REDIS_DSN', 'redis://localhost:6379')
REDIS_SETTINGS = RedisSettings.from_dsn(REDIS_DSN)

async def download_content(ctx, url: str) -> int:
    """An example job function that downloads content from a URL."""
    session: httpx.AsyncClient = ctx['session']
    response = await session.get(url, timeout=5) # Added timeout for robustness
    response.raise_for_status()
    print(f"Worker: Downloaded {len(response.text)} bytes from {url[:50]}...")
    return len(response.text)

async def startup(ctx):
    """Worker startup hook to initialize shared resources."""
    print("Worker: Starting up - creating httpx.AsyncClient session.")
    ctx['session'] = httpx.AsyncClient()

async def shutdown(ctx):
    """Worker shutdown hook to clean up shared resources."""
    if 'session' in ctx:
        await ctx['session'].aclose()
        print("Worker: Shutting down - closing httpx.AsyncClient session.")

class MyWorkerSettings(WorkerSettings):
    """Worker settings for arq CLI."""
    functions = [download_content]
    on_startup = startup
    on_shutdown = shutdown
    redis_settings = REDIS_SETTINGS
    keep_result = 60 * 60 # Keep job results for 1 hour

async def main():
    """Producer: Enqueues jobs."""
    print(f"Producer: Connecting to Redis at {REDIS_DSN}")
    redis = await create_pool(REDIS_SETTINGS)
    
    urls = [
        'https://www.google.com',
        'https://www.bing.com',
        'https://www.yahoo.com',
    ]

    for url in urls:
        job = await redis.enqueue_job('download_content', url)
        print(f"Producer: Enqueued job {job.job_id} for {url}")
    
    # Optional: Wait for a job result (for demonstration)
    # first_job_id = (await redis.queued_jobs())[0].job_id if await redis.queued_jobs() else None
    # if first_job_id:
    #     job = await redis.job(first_job_id)
    #     print(f"Producer: Waiting for job {job.job_id} result...")
    #     result = await job.result(timeout=10) # Wait up to 10 seconds
    #     print(f"Producer: Job {job.job_id} finished with result: {result}")
    
    redis.close()
    await redis.wait_closed()
    print("Producer: Redis connection closed.")

if __name__ == '__main__':
    # To run the producer (enqueue jobs):
    # python your_script_name.py
    asyncio.run(main())

    # To run the worker (in a separate terminal):
    # ARQ_REDIS_DSN='redis://localhost:6379' arq your_script_name.MyWorkerSettings --burst
    # (Remove --burst to run continuously)