Minique

raw JSON →

0.11.0 verified Fri May 01 auth: no python

Minimal Redis job runner for Python. Version 0.11.0, released sporadically. Requires Python >=3.9, depends on Redis.

pip install minique[redis]

Common errors

error TypeError: enqueue() missing 1 required positional argument: 'queue' ↓

cause Calling enqueue without a Queue instance.

fix

Pass a Queue object: enqueue(queue, function_name).

error ModuleNotFoundError: No module named 'minique.enqueue' ↓

cause Old import path after 0.10.0 breaking change.

fix

Use from minique import enqueue.

error redis.exceptions.ConnectionError: Error 111 connecting to localhost:6379. Connection refused. ↓

cause Redis server not running or wrong REDIS_URL.

fix

Start Redis (e.g., redis-server) or set REDIS_URL environment variable.

Warnings

breaking In 0.10.0, the `enqueue` function was moved from `minique.enqueue` to `minique` top-level. Old import `from minique.enqueue import enqueue` breaks. ↓

fix Use `from minique import enqueue`.

gotcha The job function must be importable by the worker. If using a dotted path string, ensure the module is in the same Python path. ↓

fix Use absolute import paths, e.g., 'mymodule.myfunc'.

deprecated The `minique.Worker` class is deprecated in 0.11.0; use the `worker()` convenience function instead. ↓

fix Replace `Worker(queue).run()` with `worker(queue)`.

Imports

Queue

wrong

from minique.queue import Queue

correct

from minique import Queue

Queue is exported from top-level package

Job

wrong

from minique.job import Job

correct

from minique import Job

Job is exported from top-level package

enqueue
```
from minique import enqueue
```
worker
```
from minique import worker
```

Quickstart

Creates a Redis-backed job queue, enqueues a function call, and processes jobs in burst mode.

import os
import redis
from minique import Queue, enqueue, worker

redis_client = redis.Redis.from_url(os.environ.get('REDIS_URL', 'redis://localhost:6379/0'))
queue = Queue('default', connection=redis_client)

# Enqueue a job
job = enqueue(queue, 'my_function', args=(1, 2), kwargs={})
print(f'Enqueued job {job.id}')

# Work off jobs
worker(queue, burst=True)