# TaskIQ-Redis
Taskiq-redis is a plugin for taskiq that adds a new broker and result backend based on redis.
# Installation
To use this project you must have installed core taskiq library:
```bash
pip install taskiq
```
This project can be installed using pip:
```bash
pip install taskiq-redis
```
# Usage
Let's see the example with the redis broker and redis async result:
```python
# broker.py
import asyncio
from taskiq_redis import ListQueueBroker, RedisAsyncResultBackend
redis_async_result = RedisAsyncResultBackend(
redis_url="redis://localhost:6379",
)
# Or you can use PubSubBroker if you need broadcasting
broker = ListQueueBroker(
url="redis://localhost:6379",
result_backend=redis_async_result,
)
@broker.task
async def best_task_ever() -> None:
"""Solve all problems in the world."""
await asyncio.sleep(5.5)
print("All problems are solved!")
async def main():
task = await best_task_ever.kiq()
print(await task.wait_result())
if __name__ == "__main__":
asyncio.run(main())
```
Launch the workers:
`taskiq worker broker:broker`
Then run the main code:
`python3 broker.py`
## PubSubBroker and ListQueueBroker configuration
We have two brokers with similar interfaces, but with different logic.
The PubSubBroker uses redis' pubsub mechanism and is very powerful,
but it executes every task on all workers, because PUBSUB broadcasts message
to all subscribers.
If you want your messages to be processed only once, please use ListQueueBroker.
It uses redis' [LPUSH](https://redis.io/commands/lpush/) and [BRPOP](https://redis.io/commands/brpop/) commands to deal with messages.
Brokers parameters:
* `url` - url to redis.
* `task_id_generator` - custom task_id genertaor.
* `result_backend` - custom result backend.
* `queue_name` - name of the pub/sub channel in redis.
* `max_connection_pool_size` - maximum number of connections in pool.
## RedisAsyncResultBackend configuration
RedisAsyncResultBackend parameters:
* `redis_url` - url to redis.
* `keep_results` - flag to not remove results from Redis after reading.
* `result_ex_time` - expire time in seconds (by default - not specified)
* `result_px_time` - expire time in milliseconds (by default - not specified)
> IMPORTANT: **It is highly recommended to use expire time in RedisAsyncResultBackend**
> If you want to add expiration, either `result_ex_time` or `result_px_time` must be set.
>```python
># First variant
>redis_async_result = RedisAsyncResultBackend(
> redis_url="redis://localhost:6379",
> result_ex_time=1000,
>)
>
># Second variant
>redis_async_result = RedisAsyncResultBackend(
> redis_url="redis://localhost:6379",
> result_px_time=1000000,
>)
>```
Raw data
{
"_id": null,
"home_page": "https://github.com/taskiq-python/taskiq-redis",
"name": "taskiq-redis",
"maintainer": "",
"docs_url": null,
"requires_python": ">=3.8.1,<4.0.0",
"maintainer_email": "",
"keywords": "taskiq,tasks,distributed,async,redis,result_backend",
"author": "taskiq-team",
"author_email": "taskiq@norely.com",
"download_url": "https://files.pythonhosted.org/packages/7e/a6/0050c41b68e74c6733e45f3955fd7e87df073ea53b2dc29cd44f3ef3c3e5/taskiq_redis-0.5.5.tar.gz",
"platform": null,
"description": "# TaskIQ-Redis\n\nTaskiq-redis is a plugin for taskiq that adds a new broker and result backend based on redis.\n\n# Installation\n\nTo use this project you must have installed core taskiq library:\n```bash\npip install taskiq\n```\nThis project can be installed using pip:\n```bash\npip install taskiq-redis\n```\n\n# Usage\n\nLet's see the example with the redis broker and redis async result:\n\n```python\n# broker.py\nimport asyncio\n\nfrom taskiq_redis import ListQueueBroker, RedisAsyncResultBackend\n\nredis_async_result = RedisAsyncResultBackend(\n redis_url=\"redis://localhost:6379\",\n)\n\n# Or you can use PubSubBroker if you need broadcasting\nbroker = ListQueueBroker(\n url=\"redis://localhost:6379\",\n result_backend=redis_async_result,\n)\n\n\n@broker.task\nasync def best_task_ever() -> None:\n \"\"\"Solve all problems in the world.\"\"\"\n await asyncio.sleep(5.5)\n print(\"All problems are solved!\")\n\n\nasync def main():\n task = await best_task_ever.kiq()\n print(await task.wait_result())\n\n\nif __name__ == \"__main__\":\n asyncio.run(main())\n```\n\nLaunch the workers:\n`taskiq worker broker:broker`\nThen run the main code:\n`python3 broker.py`\n\n## PubSubBroker and ListQueueBroker configuration\n\nWe have two brokers with similar interfaces, but with different logic.\nThe PubSubBroker uses redis' pubsub mechanism and is very powerful,\nbut it executes every task on all workers, because PUBSUB broadcasts message\nto all subscribers.\n\nIf you want your messages to be processed only once, please use ListQueueBroker.\nIt uses redis' [LPUSH](https://redis.io/commands/lpush/) and [BRPOP](https://redis.io/commands/brpop/) commands to deal with messages.\n\nBrokers parameters:\n* `url` - url to redis.\n* `task_id_generator` - custom task_id genertaor.\n* `result_backend` - custom result backend.\n* `queue_name` - name of the pub/sub channel in redis.\n* `max_connection_pool_size` - maximum number of connections in pool.\n\n## RedisAsyncResultBackend configuration\n\nRedisAsyncResultBackend parameters:\n* `redis_url` - url to redis.\n* `keep_results` - flag to not remove results from Redis after reading.\n* `result_ex_time` - expire time in seconds (by default - not specified)\n* `result_px_time` - expire time in milliseconds (by default - not specified)\n> IMPORTANT: **It is highly recommended to use expire time \u200b\u200bin RedisAsyncResultBackend**\n> If you want to add expiration, either `result_ex_time` or `result_px_time` must be set.\n>```python\n># First variant\n>redis_async_result = RedisAsyncResultBackend(\n> redis_url=\"redis://localhost:6379\",\n> result_ex_time=1000,\n>)\n>\n># Second variant\n>redis_async_result = RedisAsyncResultBackend(\n> redis_url=\"redis://localhost:6379\",\n> result_px_time=1000000,\n>)\n>```\n",
"bugtrack_url": null,
"license": "",
"summary": "Redis integration for taskiq",
"version": "0.5.5",
"project_urls": {
"Homepage": "https://github.com/taskiq-python/taskiq-redis",
"Repository": "https://github.com/taskiq-python/taskiq-redis"
},
"split_keywords": [
"taskiq",
"tasks",
"distributed",
"async",
"redis",
"result_backend"
],
"urls": [
{
"comment_text": "",
"digests": {
"blake2b_256": "61900795ff340d21bbec37e2dbc80b62185531fba4d41348f1be2549b8a1b094",
"md5": "2656a34ac8a6325158a3a36e3c3db7d5",
"sha256": "4a7add1e30b13bac771c21caf7217f79a417fd28ebb0d33bdca67b9ebb76bd23"
},
"downloads": -1,
"filename": "taskiq_redis-0.5.5-py3-none-any.whl",
"has_sig": false,
"md5_digest": "2656a34ac8a6325158a3a36e3c3db7d5",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.8.1,<4.0.0",
"size": 8687,
"upload_time": "2023-12-12T09:46:39",
"upload_time_iso_8601": "2023-12-12T09:46:39.022101Z",
"url": "https://files.pythonhosted.org/packages/61/90/0795ff340d21bbec37e2dbc80b62185531fba4d41348f1be2549b8a1b094/taskiq_redis-0.5.5-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": "",
"digests": {
"blake2b_256": "7ea60050c41b68e74c6733e45f3955fd7e87df073ea53b2dc29cd44f3ef3c3e5",
"md5": "7f359cfba327286dd46b44d6f97a1ce4",
"sha256": "251f0af753de2742128d6e3a3beed40cf9222537596882b604c0f85353c2b5a0"
},
"downloads": -1,
"filename": "taskiq_redis-0.5.5.tar.gz",
"has_sig": false,
"md5_digest": "7f359cfba327286dd46b44d6f97a1ce4",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.8.1,<4.0.0",
"size": 7840,
"upload_time": "2023-12-12T09:46:40",
"upload_time_iso_8601": "2023-12-12T09:46:40.955106Z",
"url": "https://files.pythonhosted.org/packages/7e/a6/0050c41b68e74c6733e45f3955fd7e87df073ea53b2dc29cd44f3ef3c3e5/taskiq_redis-0.5.5.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2023-12-12 09:46:40",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "taskiq-python",
"github_project": "taskiq-redis",
"travis_ci": false,
"coveralls": false,
"github_actions": true,
"lcname": "taskiq-redis"
}
JSON
