bullmq-extended (2.25.4)
Installation
pip install --index-url About this package
BullMQ for Python
BullMQ For Python
This is the official BullMQ Python library. It is a close port of the NodeJS version of the library. Python Queues are interoperable with NodeJS Queues, as both libraries use the same .lua scripts that power all the functionality.
Features
Currently, the library does not support all the features available in the NodeJS version. The following have been ported so far:
-
Add jobs to queues.
- Regular jobs.
- Delayed jobs.
- Job deduplication.
- Job priority.
- Repeatable job schedulers with
everyintervals. - Repeatable job schedulers with cron patterns.
-
Workers
-
Job events.
-
Job progress.
-
Job retries.
-
Job backoff.
-
Getters.
Installation
pip install bullmq
Usage
Basic Example
from bullmq import Queue
queue = Queue('my-queue')
job = await queue.add('my-job', {'foo': 'bar'})
Queue Events
from bullmq import QueueEvents
queue_events = QueueEvents('my-queue')
queue_events.on('completed', lambda args, event_id: print(args['jobId']))
queue_events.on('failed', lambda args, event_id: print(args['failedReason']))
Event callbacks are invoked synchronously by the QueueEvents consumer. If a
callback returns a coroutine, it is scheduled as an asyncio task; exceptions from
synchronous callbacks or scheduled coroutine callbacks are emitted as error
events and do not stop event consumption.
Custom Queue Events
from bullmq import QueueEvents, QueueEventsProducer
queue_events = QueueEvents('my-queue')
producer = QueueEventsProducer('my-queue')
queue_events.on('custom-event', lambda args, event_id: print(args))
await producer.publish_event('custom-event', {'jobId': 'custom-1'})
Wait For a Job Result
from bullmq import Queue, QueueEvents
queue = Queue('my-queue')
queue_events = QueueEvents('my-queue')
job = await queue.add('my-job', {'foo': 'bar'})
result = await job.wait_until_finished(queue_events, ttl=30000)
wait_until_finished should share one QueueEvents instance across many
waiters for the same queue. If a job uses removeOnComplete and the completion
event is missed, the Python client cannot recover the result after the job hash
has been removed. Keep completed-job result retention at least as long as the
maximum wait timeout used by callers that need the return value.
Manual Cross-Queue Orchestration
import asyncio
from bullmq import Queue, QueueEvents
image_queue = Queue('image-work')
email_queue = Queue('email-work')
image_events = QueueEvents('image-work')
email_events = QueueEvents('email-work')
image_job = await image_queue.add('resize', {'image_id': 'img_123'})
email_job = await email_queue.add('draft', {'user_id': 'usr_123'})
image_result, email_result = await asyncio.gather(
image_job.wait_until_finished(image_events, ttl=60000),
email_job.wait_until_finished(email_events, ttl=60000),
)
Queue Stats and Prometheus Metrics
from bullmq import Queue
queue = Queue('my-queue')
stats = await queue.get_stats()
histogram = await queue.get_histogram('process_duration')
queues = await Queue.get_queues({'prefix': 'bull'})
metrics = await Queue.get_prometheus_metrics({'prefix': 'bull'})
get_job_counts() reports current retained queue state, so completed or failed
counts can shrink when removeOnComplete or removeOnFail removes jobs.
get_stats() reports durable completed/failed totals from queue-local stats
keys and includes current operational state gauges for waiting, active,
delayed, prioritized, waiting-children, completed, failed, and
paused.
The older get_metrics() API still returns BullMQ's minute-bucket time series
when worker metrics.maxDataPoints is configured. Queue stats and histograms are
updated by Lua scripts by default and do not require a separate collector.
Job Schedulers
from bullmq import Queue
queue = Queue('reports')
job = await queue.upsert_job_scheduler(
'daily-summary',
{'every': 24 * 60 * 60 * 1000},
{'name': 'summary', 'data': {'kind': 'daily'}},
)
Job Deduplication
Prevent duplicate jobs from being added to the queue:
from bullmq import Queue
queue = Queue('my-queue')
# Simple mode - deduplicates until job completes or fails
job = await queue.add('paint', {'color': 'white'}, {
'deduplication': {
'id': 'custom-dedup-id'
}
})
# Throttle mode - deduplicates for a specific time window (in milliseconds)
job = await queue.add('paint', {'color': 'white'}, {
'deduplication': {
'id': 'custom-dedup-id',
'ttl': 5000 # 5 seconds
}
})
# Debounce mode - replaces pending job with latest data
job = await queue.add('paint', {'color': 'white'}, {
'deduplication': {
'id': 'custom-dedup-id',
'ttl': 5000,
'extend': True, # Extend TTL on each duplicate attempt
'replace': True # Replace job data with latest
},
'delay': 5000 # Must be delayed for replace to work
})
Documentation
The documentation is available at https://docs.bullmq.io
License
MIT
Copyright
Copyright (c) 2018-2023, Taskforce.sh Inc. and other contributors.