# Task Queue
Task Queue is an advanced queue system for Beanie (MongoDB), designed to efficiently manage and process tasks. It features task priorities, states, dependencies, and automatic expiration. Different task queues can be processed together using the Worker class. Multiple workers can be run in separate processes using the Runner class.
## Installation
```shell
pip install beanie[queue]
```
## Example
```python
from beanie_batteries_queue import Task, Runner
class ExampleTask(Task):
data: str
async def run(self):
self.data = self.data.upper()
await self.save()
runner = Runner(task_classes=[ExampleTask])
runner.start()
```
## Task
### Declare a task class
```python
from beanie_batteries_queue import Task
class SimpleTask(Task):
s: str
```
### Process a task
```python
from beanie_batteries_queue import State
# Producer
task = SimpleTask(s="test")
await task.push()
# Consumer
async for task in SimpleTask.queue():
assert task.s == "test"
# Do some work
await task.finish()
break
# Check that the task is finished
task = await SimpleTask.find_one({"s": "test"})
assert task.state == State.FINISHED
```
Async generator `SimpleTask.queue()` will return all unfinished tasks in the order they were created or based on the
priority if it was specified. It is an infinite loop, so you can use `break` to stop it.
You can also use `SimpleTask.pop()` to get the next task from the queue.
```python
from beanie_batteries_queue import State
# Producer
task = SimpleTask(s="test")
await task.push()
# Consumer
task = await SimpleTask.pop()
assert task.s == "test"
# Do some work
await task.finish()
```
### Task priority
There are three priority levels: `LOW`, `MEDIUM`, and `HIGH`. The default priority is `MEDIUM`.
Tasks are popped from the queue in the following order: `HIGH`, `MEDIUM`, `LOW`.
```python
from beanie_batteries_queue import Priority
task1 = SimpleTask(s="test1", priority=Priority.LOW)
await task1.push()
task2 = SimpleTask(s="test2", priority=Priority.HIGH)
await task2.push()
async for task in SimpleTask.queue():
assert task.s == "test2"
await task.finish()
break
```
### Task state
There are four states: `CREATED`, `RUNNING`, `FINISHED`, and `FAILED`. The default state is `PENDING`.
When a task is pushed, it is in the `CREATED` state. When it gets popped from the queue, it is in the `RUNNING`
state. `FINISHED` and `FAILED` states should be set manually.
Finished:
```python
from beanie_batteries_queue import State
task = SimpleTask(s="test")
await task.push()
async for task in SimpleTask.queue():
assert task.state == State.RUNNING
await task.finish()
break
task = await SimpleTask.find_one({"s": "test"})
assert task.state == State.FINISHED
```
Failed:
```python
from beanie_batteries_queue import State
task = SimpleTask(s="test")
await task.push()
async for task in SimpleTask.queue():
assert task.state == State.RUNNING
await task.fail()
break
task = await SimpleTask.find_one({"s": "test"})
assert task.state == State.FAILED
```
### Task dependencies
You can specify that a task depends on another task. In this case, the task will be popped from the queue only when all
its dependencies have finished.
```python
from beanie_batteries_queue import Task, DependencyType
from beanie_batteries_queue import Link
from pydantic import Field
class SimpleTask(Task):
s: str
class TaskWithDirectDependency(Task):
s: str
direct_dependency: Link[SimpleTask] = Field(
dependency_type=DependencyType.DIRECT
)
```
```python
from beanie_batteries_queue import State
task1 = SimpleTask(s="test1")
await task1.push()
task2 = TaskWithDirectDependency(s="test2", direct_dependency=task1)
await task2.push()
task_from_queue = await TaskWithDirectDependency.pop()
assert task_from_queue is None
# task2 is not popped from the queue because task1 is not finished yet
await task1.finish()
task_from_queue = await TaskWithDirectDependency.pop()
assert task_from_queue is not None
# task2 is popped from the queue because task1 is finished
```
### Task dependencies with multiple links
You can specify that a task depends on multiple tasks. In this case, the task will be popped from the queue when all or
any its dependencies are finished. It is controlled by the `dependency_type` parameter.
All
```python
class TaskWithMultipleDependencies(Task):
s: str
list_of_dependencies: Link[SimpleTask] = Field(
dependency_type=DependencyType.ALL_OF
)
```
Any
```python
class TaskWithMultipleDependencies(Task):
s: str
list_of_dependencies: Link[SimpleTask] = Field(
dependency_type=DependencyType.ANY_OF
)
```
Tasks can have multiple links with different dependency types.
```python
class TaskWithMultipleDependencies(Task):
s: str
list_of_dependencies_all: Link[SimpleTask] = Field(
dependency_type=DependencyType.ALL_OF
)
list_of_dependencies_any: Link[SimpleTask] = Field(
dependency_type=DependencyType.ANY_OF
)
direct_dependency: Link[SimpleTask] = Field(
dependency_type=DependencyType.DIRECT
)
```
### Expire time
You can specify the time after which the task will be removed from the queue, even if it is not finished or has failed.
This is controlled by the `expireAfterSeconds` index, which is set to 24 hours by default.
```python
from pymongo import ASCENDING
from beanie_batteries_queue import Task
class TaskWithExpireTime(Task):
s: str
class Settings:
indexes = [
# Other indexes,
# Expire after 5 minutes
[("created_at", ASCENDING), ("expireAfterSeconds", 300)],
]
```
Finished or failed tasks are not immediately removed from the queue. They are removed after the expiration time. You can
manually delete them using the `delete()` method.
## Queue
Queues are designed to manage tasks. It will handle all the logic of creating, updating, and deleting tasks. Task logic
should be defined in the `run` method of the task
```python
from beanie_batteries_queue import Task
class ProcessTask(Task):
data: str
async def run(self):
# Implement the logic for processing the task
print(f"Processing task with data: {self.data}")
self.data = self.data.upper()
await self.save()
```
Now we can start the queue and it will process all the tasks. Be aware - it will run infinite loop. If you want to have
another logic after starting the queue, you should run it with `asyncio.create_task()`.
```python
queue = ProcessTask.queue()
await queue.start()
```
### Stop the queue
You can stop the queue by calling the `stop()` method.
```python
await queue.stop()
```
### Queue settings
You can specify how frequently the queue will check for new tasks. The default value is 1 second.
```python
queue = ProcessTask.queue(sleep_time=60) # 60 seconds
await queue.start()
```
## Worker
Queue can handle only one task model. To process multiple task models, you should use Worker. It will run multiple queues
```python
from beanie_batteries_queue import Task, Worker
class ProcessTask(Task):
data: str
async def run(self):
self.data = self.data.upper()
await self.save()
class AnotherTask(Task):
data: str
async def run(self):
self.data = self.data.upper()
await self.save()
worker = Worker(task_classes=[ProcessTask, AnotherTask])
await worker.start()
```
Be aware - it will run infinite loop. If you want to have another logic after starting the worker, you should run it with `asyncio.create_task()`.
### Stop the worker
You can stop the worker by calling the `stop()` method.
```python
await worker.stop()
```
### Worker settings
You can specify how frequently the worker will check for new tasks. The default value is 1 second.
```python
worker = Worker(task_classes=[ProcessTask, AnotherTask], sleep_time=60) # 60 seconds
await worker.start()
```
## Runner
Runner is a class that allows you to run multiple workers in separate processes. It is useful when your tasks are CPU intensive and you want to use all the cores of your CPU.
```python
from beanie_batteries_queue import Task, Runner
class ProcessTask(Task):
data: str
async def run(self):
self.data = self.data.upper()
await self.save()
class AnotherTask(Task):
data: str
async def run(self):
self.data = self.data.upper()
await self.save()
runner = Runner(task_classes=[ProcessTask, AnotherTask])
runner.start()
```
### Stop the runner
You can stop the runner by calling the `stop()` method.
```python
runner.stop()
```
### Runner settings
You can specify how many workers will be run. The default value is 1.
```python
runner = Runner(task_classes=[ProcessTask, AnotherTask], workers_count=4)
runner.start()
```
You can specify how frequently the worker will check for new tasks. The default value is 1 second.
```python
runner = Runner(task_classes=[ProcessTask, AnotherTask], sleep_time=60) # 60 seconds
runner.start()
```
You can specify if the start method should run while the workers are alive or if it should return immediately. The default value is True.
```python
runner = Runner(task_classes=[ProcessTask, AnotherTask], run_indefinitely=False)
runner.start()
```
Raw data
{
"_id": null,
"home_page": null,
"name": "beanie_batteries_queue",
"maintainer": null,
"docs_url": null,
"requires_python": ">=3.7,<4.0",
"maintainer_email": null,
"keywords": "mongodb,odm,orm,pydantic,mongo,async,python,beanie,queue,beanie-batteries-queue",
"author": null,
"author_email": "Roman Right <roman-right@protonmail.com>",
"download_url": "https://files.pythonhosted.org/packages/d3/88/a9eb9c6f99e0f6c5443b7ac8d4396db76ca250f2c4d289f5ddb9ad2606c5/beanie_batteries_queue-0.4.0.tar.gz",
"platform": null,
"description": "# Task Queue\n\nTask Queue is an advanced queue system for Beanie (MongoDB), designed to efficiently manage and process tasks. It features task priorities, states, dependencies, and automatic expiration. Different task queues can be processed together using the Worker class. Multiple workers can be run in separate processes using the Runner class.\n\n## Installation\n\n```shell\npip install beanie[queue]\n```\n\n## Example\n\n```python\nfrom beanie_batteries_queue import Task, Runner\n\nclass ExampleTask(Task):\n data: str\n\n async def run(self):\n self.data = self.data.upper()\n await self.save()\n \nrunner = Runner(task_classes=[ExampleTask])\nrunner.start()\n```\n\n## Task\n\n### Declare a task class\n\n```python\nfrom beanie_batteries_queue import Task\n\n\nclass SimpleTask(Task):\n s: str\n```\n\n### Process a task\n\n```python\nfrom beanie_batteries_queue import State\n\n# Producer\ntask = SimpleTask(s=\"test\")\nawait task.push()\n\n# Consumer\nasync for task in SimpleTask.queue():\n assert task.s == \"test\"\n # Do some work\n await task.finish()\n break\n\n# Check that the task is finished\ntask = await SimpleTask.find_one({\"s\": \"test\"})\nassert task.state == State.FINISHED\n```\n\nAsync generator `SimpleTask.queue()` will return all unfinished tasks in the order they were created or based on the\npriority if it was specified. It is an infinite loop, so you can use `break` to stop it.\n\nYou can also use `SimpleTask.pop()` to get the next task from the queue.\n\n```python\nfrom beanie_batteries_queue import State\n\n# Producer\ntask = SimpleTask(s=\"test\")\nawait task.push()\n\n# Consumer\ntask = await SimpleTask.pop()\nassert task.s == \"test\"\n# Do some work\nawait task.finish()\n```\n\n### Task priority\n\nThere are three priority levels: `LOW`, `MEDIUM`, and `HIGH`. The default priority is `MEDIUM`.\nTasks are popped from the queue in the following order: `HIGH`, `MEDIUM`, `LOW`.\n\n```python\nfrom beanie_batteries_queue import Priority\n\ntask1 = SimpleTask(s=\"test1\", priority=Priority.LOW)\nawait task1.push()\ntask2 = SimpleTask(s=\"test2\", priority=Priority.HIGH)\nawait task2.push()\n\nasync for task in SimpleTask.queue():\n assert task.s == \"test2\"\n await task.finish()\n break\n```\n\n### Task state\n\nThere are four states: `CREATED`, `RUNNING`, `FINISHED`, and `FAILED`. The default state is `PENDING`.\nWhen a task is pushed, it is in the `CREATED` state. When it gets popped from the queue, it is in the `RUNNING`\nstate. `FINISHED` and `FAILED` states should be set manually.\n\nFinished:\n\n```python\nfrom beanie_batteries_queue import State\n\ntask = SimpleTask(s=\"test\")\nawait task.push()\n\nasync for task in SimpleTask.queue():\n assert task.state == State.RUNNING\n await task.finish()\n break\n\ntask = await SimpleTask.find_one({\"s\": \"test\"})\nassert task.state == State.FINISHED\n```\n\nFailed:\n\n```python\nfrom beanie_batteries_queue import State\n\ntask = SimpleTask(s=\"test\")\nawait task.push()\n\nasync for task in SimpleTask.queue():\n assert task.state == State.RUNNING\n await task.fail()\n break\n\ntask = await SimpleTask.find_one({\"s\": \"test\"})\nassert task.state == State.FAILED\n```\n\n### Task dependencies\n\nYou can specify that a task depends on another task. In this case, the task will be popped from the queue only when all\nits dependencies have finished.\n\n```python\nfrom beanie_batteries_queue import Task, DependencyType\nfrom beanie_batteries_queue import Link\nfrom pydantic import Field\n\n\nclass SimpleTask(Task):\n s: str\n\n\nclass TaskWithDirectDependency(Task):\n s: str\n direct_dependency: Link[SimpleTask] = Field(\n dependency_type=DependencyType.DIRECT\n )\n```\n\n```python\nfrom beanie_batteries_queue import State\n\ntask1 = SimpleTask(s=\"test1\")\nawait task1.push()\n\ntask2 = TaskWithDirectDependency(s=\"test2\", direct_dependency=task1)\nawait task2.push()\n\ntask_from_queue = await TaskWithDirectDependency.pop()\nassert task_from_queue is None\n# task2 is not popped from the queue because task1 is not finished yet\n\nawait task1.finish()\n\ntask_from_queue = await TaskWithDirectDependency.pop()\nassert task_from_queue is not None\n# task2 is popped from the queue because task1 is finished\n```\n\n### Task dependencies with multiple links\n\nYou can specify that a task depends on multiple tasks. In this case, the task will be popped from the queue when all or\nany its dependencies are finished. It is controlled by the `dependency_type` parameter.\n\nAll\n\n```python\nclass TaskWithMultipleDependencies(Task):\n s: str\n list_of_dependencies: Link[SimpleTask] = Field(\n dependency_type=DependencyType.ALL_OF\n )\n```\n\nAny\n\n```python\nclass TaskWithMultipleDependencies(Task):\n s: str\n list_of_dependencies: Link[SimpleTask] = Field(\n dependency_type=DependencyType.ANY_OF\n )\n```\n\nTasks can have multiple links with different dependency types.\n\n```python\nclass TaskWithMultipleDependencies(Task):\n s: str\n list_of_dependencies_all: Link[SimpleTask] = Field(\n dependency_type=DependencyType.ALL_OF\n )\n list_of_dependencies_any: Link[SimpleTask] = Field(\n dependency_type=DependencyType.ANY_OF\n )\n direct_dependency: Link[SimpleTask] = Field(\n dependency_type=DependencyType.DIRECT\n )\n```\n\n### Expire time\n\nYou can specify the time after which the task will be removed from the queue, even if it is not finished or has failed.\nThis is controlled by the `expireAfterSeconds` index, which is set to 24 hours by default.\n\n```python\nfrom pymongo import ASCENDING\nfrom beanie_batteries_queue import Task\n\n\nclass TaskWithExpireTime(Task):\n s: str\n\n class Settings:\n indexes = [\n # Other indexes,\n\n # Expire after 5 minutes\n [(\"created_at\", ASCENDING), (\"expireAfterSeconds\", 300)],\n ]\n```\n\nFinished or failed tasks are not immediately removed from the queue. They are removed after the expiration time. You can\nmanually delete them using the `delete()` method.\n\n## Queue\n\nQueues are designed to manage tasks. It will handle all the logic of creating, updating, and deleting tasks. Task logic\nshould be defined in the `run` method of the task\n\n```python\nfrom beanie_batteries_queue import Task\n\n\nclass ProcessTask(Task):\n data: str\n\n async def run(self):\n # Implement the logic for processing the task\n print(f\"Processing task with data: {self.data}\")\n self.data = self.data.upper()\n await self.save()\n```\n\nNow we can start the queue and it will process all the tasks. Be aware - it will run infinite loop. If you want to have\nanother logic after starting the queue, you should run it with `asyncio.create_task()`.\n\n```python\nqueue = ProcessTask.queue()\nawait queue.start()\n```\n\n### Stop the queue\n\nYou can stop the queue by calling the `stop()` method.\n\n```python\nawait queue.stop()\n```\n\n### Queue settings\n\nYou can specify how frequently the queue will check for new tasks. The default value is 1 second.\n\n```python\nqueue = ProcessTask.queue(sleep_time=60) # 60 seconds\nawait queue.start()\n```\n## Worker\n\nQueue can handle only one task model. To process multiple task models, you should use Worker. It will run multiple queues\n\n```python\nfrom beanie_batteries_queue import Task, Worker\n\nclass ProcessTask(Task):\n data: str\n\n async def run(self):\n self.data = self.data.upper()\n await self.save()\n\nclass AnotherTask(Task):\n data: str\n\n async def run(self):\n self.data = self.data.upper()\n await self.save()\n \n\nworker = Worker(task_classes=[ProcessTask, AnotherTask])\nawait worker.start()\n```\n\nBe aware - it will run infinite loop. If you want to have another logic after starting the worker, you should run it with `asyncio.create_task()`.\n\n### Stop the worker\n\nYou can stop the worker by calling the `stop()` method.\n\n```python\nawait worker.stop()\n```\n\n### Worker settings\n\nYou can specify how frequently the worker will check for new tasks. The default value is 1 second.\n\n```python\nworker = Worker(task_classes=[ProcessTask, AnotherTask], sleep_time=60) # 60 seconds\nawait worker.start()\n```\n\n## Runner\n\nRunner is a class that allows you to run multiple workers in separate processes. It is useful when your tasks are CPU intensive and you want to use all the cores of your CPU.\n\n```python\nfrom beanie_batteries_queue import Task, Runner\n\nclass ProcessTask(Task):\n data: str\n\n async def run(self):\n self.data = self.data.upper()\n await self.save()\n\nclass AnotherTask(Task):\n data: str\n\n async def run(self):\n self.data = self.data.upper()\n await self.save()\n\nrunner = Runner(task_classes=[ProcessTask, AnotherTask])\nrunner.start()\n```\n\n### Stop the runner\n\nYou can stop the runner by calling the `stop()` method.\n\n```python\nrunner.stop()\n```\n\n### Runner settings\n\nYou can specify how many workers will be run. The default value is 1.\n\n```python\nrunner = Runner(task_classes=[ProcessTask, AnotherTask], workers_count=4)\nrunner.start()\n```\n\nYou can specify how frequently the worker will check for new tasks. The default value is 1 second.\n\n```python\nrunner = Runner(task_classes=[ProcessTask, AnotherTask], sleep_time=60) # 60 seconds\nrunner.start()\n```\n\nYou can specify if the start method should run while the workers are alive or if it should return immediately. The default value is True.\n\n```python\nrunner = Runner(task_classes=[ProcessTask, AnotherTask], run_indefinitely=False)\nrunner.start()\n```\n",
"bugtrack_url": null,
"license": null,
"summary": "Advanced queue system for MongoDB with Beanie ODM",
"version": "0.4.0",
"project_urls": {
"homepage": "https://github.com/roman-right/beanie_batteries_queue",
"repository": "https://github.com/roman-right/beanie_batteries_queue"
},
"split_keywords": [
"mongodb",
"odm",
"orm",
"pydantic",
"mongo",
"async",
"python",
"beanie",
"queue",
"beanie-batteries-queue"
],
"urls": [
{
"comment_text": null,
"digests": {
"blake2b_256": "60750924b0ec8296136a948ce06eb00f32c28dfd68cf6db0c763466d22fe36e3",
"md5": "44d06b9d5d4b6cf812e9c1f8e8c3ff2c",
"sha256": "dfb5653532b1a1d6d26544fd9373d3a2e266d29bd3b612ccc0a2713f80e879a0"
},
"downloads": -1,
"filename": "beanie_batteries_queue-0.4.0-py3-none-any.whl",
"has_sig": false,
"md5_digest": "44d06b9d5d4b6cf812e9c1f8e8c3ff2c",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.7,<4.0",
"size": 13357,
"upload_time": "2023-11-13T00:51:53",
"upload_time_iso_8601": "2023-11-13T00:51:53.023547Z",
"url": "https://files.pythonhosted.org/packages/60/75/0924b0ec8296136a948ce06eb00f32c28dfd68cf6db0c763466d22fe36e3/beanie_batteries_queue-0.4.0-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": null,
"digests": {
"blake2b_256": "d388a9eb9c6f99e0f6c5443b7ac8d4396db76ca250f2c4d289f5ddb9ad2606c5",
"md5": "1962e7595416da9f02fc13c66b3245d3",
"sha256": "a5f548a2f040ca1737b239e2052316e331b491f36487578712d8e0525d2ecfcb"
},
"downloads": -1,
"filename": "beanie_batteries_queue-0.4.0.tar.gz",
"has_sig": false,
"md5_digest": "1962e7595416da9f02fc13c66b3245d3",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.7,<4.0",
"size": 18291,
"upload_time": "2023-11-13T00:51:54",
"upload_time_iso_8601": "2023-11-13T00:51:54.731652Z",
"url": "https://files.pythonhosted.org/packages/d3/88/a9eb9c6f99e0f6c5443b7ac8d4396db76ca250f2c4d289f5ddb9ad2606c5/beanie_batteries_queue-0.4.0.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2023-11-13 00:51:54",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "roman-right",
"github_project": "beanie_batteries_queue",
"travis_ci": false,
"coveralls": false,
"github_actions": true,
"lcname": "beanie_batteries_queue"
}
JSON
