Metadata-Version: 2.4
Name: bullmq-extended
Version: 2.25.6
Summary: BullMQ for Python
Author-email: "Taskforce.sh Inc." <manast@taskforce.sh>
Project-URL: Homepage, https://git.d.aiengines.ir/queue/bullmq
Project-URL: Bug Tracker, https://git.d.aiengines.ir/queue/bullmq/issues
Keywords: python,bullmq,queues
Classifier: Development Status :: 3 - Alpha
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Operating System :: POSIX :: Linux
Classifier: Programming Language :: Python :: 3.10
Requires-Python: >=3.10.0
Description-Content-Type: text/markdown
Requires-Dist: redis==7.4.0
Requires-Dist: msgpack==1.2.1
Requires-Dist: semver==3.0.4
Provides-Extra: dev
Requires-Dist: pre-commit==4.5.1; extra == "dev"
Requires-Dist: build==1.4.2; extra == "dev"
Requires-Dist: types-redis==4.6.0.20241004; extra == "dev"
Requires-Dist: pipenv==2026.5.1; extra == "dev"
Requires-Dist: virtualenv==21.2.0; extra == "dev"
Requires-Dist: virtualenv-clone==0.5.7; extra == "dev"
Provides-Extra: test
Requires-Dist: pytest==9.0.3; extra == "test"
Requires-Dist: pytest-timeout==2.4.0; extra == "test"

# 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.

  - [x] Regular jobs.
  - [x] Delayed jobs.
  - [x] Job deduplication.
  - [x] Job priority.
  - [x] Repeatable job schedulers with `every` intervals.
  - [ ] Repeatable job schedulers with cron patterns.

- [x] Workers
- [x] Job events.
- [x] Job progress.
- [x] Job retries.
- [x] Job backoff.
- [x] Getters.

## Installation

```bash
pip install --index-url https://git.d.aiengines.ir/api/packages/queue/pypi/simple bullmq-extended
```

## Usage

### Basic Example

```python
from bullmq import Queue

queue = Queue('my-queue')

job = await queue.add('my-job', {'foo': 'bar'})
```

### BullMQ Kit

BullMQ Kit provides task-style APIs on top of BullMQ queues. Pass Redis
connections the same way you do with BullMQ, including URL strings.

```python
import asyncio
from bullmq.kit import App, Context


async def main():
    app = App(connection="redis://127.0.0.1:6379")

    @app.task(name="math.add", context=True, concurrency=4, retries=2)
    async def add(ctx: Context, left: int, right: int) -> int:
        await ctx.set_progress(50)
        await ctx.log(f"Adding {left} and {right}")
        return left + right

    @app.task(name="math.multiply")
    async def multiply(value: int) -> int:
        result = await add(value, 10)
        return result * 2

    await app.connect()

    run = await multiply.start(5)
    print(await run.result())

    await app.close()


asyncio.run(main())
```

Producer-only clients can call a task registered by another process:

```python
import asyncio
from bullmq.kit import App


async def main():
    app = App(connection="redis://127.0.0.1:6379")
    add = app.task_ref("math.add")

    await app.connect()
    print(await add(10, 20))
    await app.close()


asyncio.run(main())
```

### Queue Events

```python
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

```python
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

```python
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

```python
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

```python
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

```python
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:

```python
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](https://docs.bullmq.io/python)

## License

MIT

## Copyright

Copyright (c) 2018-2023, Taskforce.sh Inc. and other contributors.
