Queue service built on top of mongo.
Project description
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.6
- Added sleep and state feature while releasing a job. This provides a way to not pickup job until provided seconds and store state for long running jobs.
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
python3.9 -m venv venv
source venv/bin/activate
pip install -r requirements.txt
pip install wheel
pip install --upgrade twine
python setup.py sdist bdist_wheel
python -m twine upload --repository-url https://upload.pypi.org/legacy/ dist/*
Local Development and Testing
cd mong_queue # Root directory of the package
python3.9 -m venv venv
source venv/bin/activate
pip install -r requirements.txt
python -m unittest mongo_queue.test
Project details
Release history Release notifications | RSS feed
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distribution
Built Distribution
Close
Hashes for mongo_queue_service-0.0.6.tar.gz
Algorithm | Hash digest | |
---|---|---|
SHA256 | 0958f806087cd7ae2660a6ea1207140f38f7516a6f8ae15b3932e0071b09f420 |
|
MD5 | a7c2c157603ff0b631bbb4f4739834c4 |
|
BLAKE2b-256 | 5ceb9f6d04d4b738612167f9a86782e4f1a99ad044a9324c8b957dd06d4329eb |
Close
Hashes for mongo_queue_service-0.0.6-py3-none-any.whl
Algorithm | Hash digest | |
---|---|---|
SHA256 | 5ba894c264500329be890ff9e4aafc921d73e9e22ee3b37ef08bd1be615c2065 |
|
MD5 | 0597d182fedc497b5d6b6b18e1118894 |
|
BLAKE2b-256 | f75dd51a6b61b9b033477ae69ad097adb09506a00aefe6cbf89115e9e6a2ee46 |