{"id":3426,"library":"bullmq","title":"BullMQ for Python","description":"BullMQ for Python is an official Python port of the popular Node.js message queue, designed for reliable background job processing using Redis. It leverages `asyncio` for efficient, concurrent task execution and is interoperable with its Node.js counterpart due to shared Lua scripts. Currently at version 2.20.3, the library sees active development with frequent bug fixes and feature enhancements, including new major versions that may introduce breaking changes.","status":"active","version":"2.20.3","language":"python","source_language":"en","source_url":"https://github.com/taskforcesh/bullmq/tree/master/python","tags":["message queue","background jobs","redis","asyncio","distributed tasks","queue"],"install":[{"cmd":"pip install bullmq","lang":"bash","label":"Install BullMQ"}],"dependencies":[{"reason":"Python Redis client for communication with the Redis server.","package":"redis","optional":false},{"reason":"MessagePack serialization for efficient data handling.","package":"msgpack","optional":false},{"reason":"Semantic versioning utilities.","package":"semver","optional":false},{"reason":"External dependency: a running Redis 5.0+ server (6.2+ recommended) is required for BullMQ to function.","package":"Redis","optional":false}],"imports":[{"symbol":"Queue","correct":"from bullmq import Queue"},{"symbol":"Worker","correct":"from bullmq import Worker"},{"symbol":"QueueEvents","correct":"from bullmq import QueueEvents"},{"symbol":"FlowProducer","correct":"from bullmq import FlowProducer"}],"quickstart":{"code":"import asyncio\nimport os\nfrom bullmq import Queue, Worker\n\nREDIS_URL = os.environ.get('REDIS_URL', 'redis://localhost:6379')\n\nasync def add_job_to_queue():\n    # Connect to Redis. Pass 'connection' as a dictionary in options.\n    queue = Queue(\"myQueue\", connection={\"host\": \"localhost\", \"port\": 6379})\n    print(f\"Adding job to queue 'myQueue' on {REDIS_URL}\")\n    job = await queue.add(\"myJobName\", {\"foo\": \"bar\"})\n    print(f\"Job added with ID: {job.id}, Data: {job.data}\")\n    await queue.close()\n\nasync def process_job(job, job_token):\n    print(f\"Processing job {job.id} with data: {job.data}\")\n    # Simulate async work\n    await asyncio.sleep(1)\n    return {\"status\": \"completed\", \"original_data\": job.data}\n\nasync def start_worker():\n    print(f\"Starting worker for queue 'myQueue' on {REDIS_URL}\")\n    # Connect to Redis. Pass 'connection' as a dictionary in options.\n    worker = Worker(\"myQueue\", process_job, connection={\"host\": \"localhost\", \"port\": 6379})\n\n    # You can listen for worker events (optional)\n    worker.on(\"completed\", lambda job, result: print(f\"Job {job.id} completed with result: {result}\"))\n    worker.on(\"failed\", lambda job, err: print(f\"Job {job.id} failed with error: {err}\"))\n\n    print(\"Worker started. Press Ctrl+C to stop.\")\n    # Keep the worker running (e.g., for a long time in a real application)\n    try:\n        while True: # Keep worker alive for demonstration\n            await asyncio.sleep(3600) # Sleep for a long time\n    except asyncio.CancelledError:\n        pass\n    finally:\n        print(\"Shutting down worker...\")\n        await worker.close()\n\nasync def main():\n    # This example assumes a Redis server is running at localhost:6379\n    # For real applications, use environment variables for connection details.\n    # docker run -d -p 6379:6379 redis:latest\n\n    # Run adding a job and starting a worker concurrently\n    # For a real application, these would typically run in separate processes/services.\n    await asyncio.gather(add_job_to_queue(), start_worker())\n\nif __name__ == \"__main__\":\n    try:\n        asyncio.run(main())\n    except KeyboardInterrupt:\n        print(\"Application stopped by user.\")\n","lang":"python","description":"This quickstart demonstrates how to add jobs to a BullMQ queue and process them with a worker. It uses `asyncio` for asynchronous operations. Ensure a Redis server is running (e.g., via `docker run -d -p 6379:6379 redis:latest`) and configure the `REDIS_URL` environment variable or provide explicit connection details. In a production environment, the queue producer and worker would typically run in separate processes or services."},"warnings":[{"fix":"Update `Queue` and `Worker` instantiation to pass connection details within the `connection` key of the options dictionary. Review application logic if directly interacting with internal Redis keys related to worker markers.","message":"BullMQ v2.0.0 introduced breaking changes. Specifically, Redis connection parameters must now be provided as part of the `options` dictionary for `Queue` and `Worker` constructors (e.g., `connection={'host': 'localhost', 'port': 6379}`). Additionally, worker markers now use a dedicated key in Redis instead of a special job ID, which impacts internal state management.","severity":"breaking","affected_versions":">=2.0.0"},{"fix":"Reduce worker concurrency, optimize job processing code, ensure stable network connectivity to Redis, consider increasing `lockDuration` in worker options, and implement proper job state checks before attempting to remove jobs. Ensure Redis `maxmemory-policy` is set to `noeviction`.","message":"Workers may encounter 'Missing lock for job X.moveToFinished' errors. This usually means a job lost its lock during processing. Common causes include high CPU usage on the worker preventing lock renewal, loss of communication with Redis, or the job being forcefully removed.","severity":"gotcha","affected_versions":"All"},{"fix":"Always define your worker's `process` function with `async def process(job, job_token): ...`.","message":"The `process` function for a BullMQ `Worker` is expected to have two positional arguments: `job` and `job_token`, even if `job_token` is not explicitly used for manual job manipulation. Omitting `job_token` will cause a `TypeError`.","severity":"gotcha","affected_versions":"All (especially in v2.x documentation examples)"},{"fix":"When creating a custom Redis connection object for BullMQ, ensure `decode_responses=True` is set if you need decoded string responses.","message":"Redis-py (the underlying client for BullMQ Python) returns binary responses by default. If you are using a custom Redis client configuration and expect string responses, you must pass `decode_responses=True` to the Redis client constructor.","severity":"gotcha","affected_versions":"All"},{"fix":"Always validate and sanitize input, especially for connection parameters, queue names, and job data. Ensure all values passed to BullMQ or its underlying Redis commands are strings or integers. Use `os.environ.get('KEY', '')` or raise errors if critical environment variables are missing.","message":"Passing undefined, empty, or non-string values (e.g., objects or arrays) when using environment variables or other dynamic inputs with BullMQ methods can lead to `ERR Error running script ... Lua redis() command arguments must be strings or integers` errors.","severity":"gotcha","affected_versions":"All"},{"fix":"Ensure that any failure condition within your `process` function explicitly raises an `Exception` (e.g., `raise ValueError('Invalid data')`). Configure `attempts` and `backoff` options when adding jobs to the queue for automatic retries.","message":"For robust error handling and retries, your worker's processor function should always raise Python `Exception` objects (or subclasses thereof) when a job fails. BullMQ relies on catching these exceptions to mark jobs as failed and apply retry logic based on `attempts` and `backoff` options.","severity":"gotcha","affected_versions":"All"}],"env_vars":null,"search_vec":"'2.20.3':49 'activ':53 'asyncio':28,76 'background':21,73 'break':69 'bug':57 'bullmq':1,4 'chang':70 'concurr':31 'counterpart':40 'current':46 'design':18 'develop':54 'distribut':77 'due':41 'effici':30 'enhanc':61 'execut':33 'featur':60 'fix':58 'frequent':56 'includ':62 'interoper':36 'introduc':68 'job':22,74 'leverag':27 'librari':51 'lua':44 'major':64 'may':67 'messag':16,71 'new':63 'node.js':15,39 'offici':9 'popular':14 'port':11 'process':23 'python':3,6,10 'queue':17,72,79 'redi':25,75 'reliabl':20 'script':45 'see':52 'share':43 'task':32,78 'use':24 'version':48,65","created_at":"2026-04-11T17:28:36.587115+00:00","updated_at":"2026-04-16T01:07:22.102064+00:00","problems":[{"fix":"Optimize your job processing code to reduce CPU load, reduce worker concurrency, ensure stable network connectivity between the worker and Redis, configure your Redis server with `maxmemory-policy noeviction`, and consider increasing the `lockDuration` in the Worker options if jobs are inherently long-running.","cause":"This error occurs when a job being processed by a worker unexpectedly loses its 'lock' in Redis, often due to high CPU usage preventing lock renewal, an unstable Redis connection, or an incorrect Redis maxmemory-policy that evicts BullMQ keys.","error":"Missing lock for job 1234. moveToFinished"},{"fix":"Install the `redis` Python client library using pip: `pip install redis`.","cause":"The official BullMQ Python library depends on the `redis` Python client, which is not automatically installed by default if you install `bullmq` directly in some environments, or if it's missing from your project's dependencies.","error":"ModuleNotFoundError: No module named 'redis'"},{"fix":"Ensure the queue name configured for the worker exactly matches the queue name used by the job producer. Verify that the Redis connection details are correct and accessible. Ensure your Python application keeps the `asyncio` event loop running for the worker, typically by using `await worker.run()` or `await asyncio.Future()` in your main worker loop. Also, ensure your processor function either completes or explicitly raises exceptions if it encounters an issue.","cause":"The BullMQ worker appears to be initialized correctly and logs that it's ready, but it does not pick up and process jobs from the queue. This can happen due to an incorrect queue name, a worker not properly connected to Redis, the main application exiting prematurely, or the processor function hanging indefinitely.","error":"Worker is ready and listening for jobs. (but jobs are not processed)"},{"fix":"Verify that your Redis server is running and accessible (e.g., by running `redis-cli ping` from your terminal, which should return `PONG`). Double-check the host, port, and any authentication credentials in your BullMQ connection configuration. Ensure no firewalls are blocking the Redis default port (6379) or your configured port.","cause":"The BullMQ client (either Queue or Worker) cannot establish a connection to the Redis server because Redis is not running, is running on a different host/port than configured, or a firewall is blocking the connection.","error":"Connection refused"},{"fix":"Adjust the signature of your worker's processor function to accept both the `job` and `token` arguments. For example: `async def process(job, token):` or `async def process(job: Job, token: Optional[str] = None):`.","cause":"This error occurs when the processor function provided to the BullMQ `Worker` is defined to accept only one argument (the `job`), but BullMQ expects it to accept two arguments (`job` and `token`) by default for manual job processing or specific API interactions.","error":"TypeError: process() takes 1 positional argument but 2 were given"}],"ecosystem":"pypi","meta_description":null,"install_score":null,"quickstart_score":null,"quickstart_tag":null,"pypi_latest":"3.1.0","cli_name":"","cli_version":null,"type":"library","homepage":"https://bullmq.io","github":"https://github.com/taskforcesh/bullmq","docs":null,"changelog":null,"pypi":"https://pypi.org/project/bullmq/","npm":null,"openapi_spec":null,"status_page":null,"smithery":null,"categories":["workflow","database"],"base_url":null,"auth_type":null,"provenance":{"verified_status":"passing","verified_at":"2026-06-28","last_verified":"2026-08-29","next_check":"2026-07-28","install_tag":null}}