mongo-queue-service 0.0.5

Queue service built on top of mongo.

mongo_queue

Task queue built on mongo with channels and unique job id.

Website • autobotAI Automation Platform

Inspired from kapilt/mongoqueue

Change Log:

v0.0.5

• Added depends_on feature. You can create dependency between jobs by supplying depends_on[] with previously created job ids.

v0.0.3

• Added unique index with job_id and channel. This is to make sure that the same job is not added multiple times. If not job id provided an unique id generated by default.

Usage

Install the package.

pip install mongo_queue

Usage Example:

• Create Queue Object

from mongo_queue.queue import Queue
from pymongo import MongoClient

queue = Queue(MongoClient('localhost', 27017).task_queue, consumer_id="consumer-1", timeout=300, max_attempts=3)

• Add task to queue default channel

queue.put({"task_id": 1})

• Add task to queue with priority to default channel

queue.put({"task_id": 1}, priority=1)

• Add task to queue in a specific channel

queue.put({"task_id": 1}, priority=1, channel="channel_1")

• Add task to queue with unique job_id

queue.put({"task_id": 1}, priority=1, channel="channel_1", job_id="x_job")

• Add task with dependency

job1 = queue.put({"task_id": 1}, priority=1, channel="channel_1", job_id="x_job")
job2 = queue.put({"task_id": 2}, priority=1, channel="channel_1", job_id="x_job", depends_on=[job1])

• Get the next job to be executed from the default channel

job = queue.next()

• Get the next job to be executed from a specific channel

job = queue.next(channel="channel_1")

• Update job progress for long-running jobs

job.progress(count=10)

• Put the job back in queue, this will be picked up again later, this will update attempts after max attempts the job will not be picked up again.

job.release()

• Put the job back in queue with error, this will be picked up again later, this will update attempts after max attempts the job will not be picked up again.

job.error("Some error occured")

• Complete the job. This will delete job from the database.

job.complete()

Build Steps

# Setup venv of python version 3.6 and above
pip install -r requirements.txt
python -m pip install --upgrade twine
python setup.py sdist bdist_wheel
python -m twine upload --repository-url https://upload.pypi.org/legacy/ dist/*