# Taskiq + Aiogram
[](https://pypi.org/project/taskiq-aiogram/)
[](https://pypi.org/project/taskiq-aiogram/)
[](https://pypistats.org/packages/taskiq-aiogram)
This repo adds integration between your aiogram application and taskiq.
It runs all startup and shutdown events of your application and adds 3 dependencies,
that you can use in your tasks.
1. `Dispatcher` - that were used along with executor;
2. `Bot` - your main bot instance;
3. `List[Bot]` - all your bots.
## Usage
Define startup and shutdown events for your dispatcher.
We use events, because we need to identify what your bot
wants to do on startup and shutdown.
Also, it's useful for bot developers to distinct buisness logic
from startup of the bot.
Below you'll find an example, how to integrate taskiq with your favorite
bot framework.
```python
import asyncio
import logging
import sys
from aiogram import Bot, Dispatcher
# Your taskiq broker
from your_project.tkq import broker
dp = Dispatcher()
bot = Bot(token="TOKEN")
bot2 = Bot(token="TOKEN")
@dp.startup()
async def setup_taskiq(bot: Bot, *_args, **_kwargs):
# Here we check if it's a clien-side,
# Becuase otherwise you're going to
# create infinite loop of startup events.
if not broker.is_worker_process:
logging.info("Setting up taskiq")
await broker.startup()
@dp.shutdown()
async def shutdown_taskiq(bot: Bot, *_args, **_kwargs):
if not broker.is_worker_process:
logging.info("Shutting down taskiq")
await broker.shutdown()
async def main():
await dp.start_polling(bot, bot2)
if __name__ == "__main__":
logging.basicConfig(level=logging.INFO, stream=sys.stdout)
asyncio.run(main())
```
The only thing that left is to add few lines to your broker definition.
```python
# Please use real broker for taskiq.
from taskiq_broker import MyBroker
import taskiq_aiogram
broker = MyBroker()
# This line is going to initialize everything.
taskiq_aiogram.init(
broker,
"your_project.__main__:dp",
"your_project.__main__:bot",
"your_project.__main__:bot2",
)
```
That's it.
Let's create some tasks! I created task in a separate module,
named `tasks.py`.
```python
from aiogram import Bot
from your_project.tkq import broker
@broker.task(task_name="my_task")
async def my_task(chat_id: int, bot: Bot = TaskiqDepends()) -> None:
print("I'm a task")
await asyncio.sleep(4)
await bot.send_message(chat_id, "task completed")
```
Now let's call our new task somewhere in bot commands.
```python
from aiogram import types
from aiogram.filters import Command
from tasks import my_task
@dp.message(Command("task"))
async def message(message: types.Message):
await my_task.kiq(message.chat.id)
```
To start the worker, please type:
```
taskiq worker your_project.tkq:broker --fs-discover
```
We use `--fs-discover` to find all tasks.py modules recursively
and import all tasks into broker.
Now we can fire the task and see everything in action.
Raw data
{
"_id": null,
"home_page": "",
"name": "taskiq-aiogram",
"maintainer": "Taskiq team",
"docs_url": null,
"requires_python": ">=3.8.1,<4.0.0",
"maintainer_email": "taskiq@no-reply.com",
"keywords": "taskiq,tasks,distributed,async",
"author": "Taskiq team",
"author_email": "taskiq@no-reply.com",
"download_url": "https://files.pythonhosted.org/packages/c3/d9/741ba505b98c1a5fd77996e3d8c1c8d57411352be305dbd8fd5e1b817234/taskiq_aiogram-0.3.1.tar.gz",
"platform": null,
"description": "# Taskiq + Aiogram\n\n[](https://pypi.org/project/taskiq-aiogram/)\n[](https://pypi.org/project/taskiq-aiogram/)\n[](https://pypistats.org/packages/taskiq-aiogram)\n\nThis repo adds integration between your aiogram application and taskiq.\n\nIt runs all startup and shutdown events of your application and adds 3 dependencies,\nthat you can use in your tasks.\n\n1. `Dispatcher` - that were used along with executor;\n2. `Bot` - your main bot instance;\n3. `List[Bot]` - all your bots.\n\n## Usage\n\nDefine startup and shutdown events for your dispatcher.\nWe use events, because we need to identify what your bot\nwants to do on startup and shutdown.\n\nAlso, it's useful for bot developers to distinct buisness logic\nfrom startup of the bot.\n\nBelow you'll find an example, how to integrate taskiq with your favorite\nbot framework.\n\n```python\nimport asyncio\nimport logging\nimport sys\n\nfrom aiogram import Bot, Dispatcher\n\n# Your taskiq broker\nfrom your_project.tkq import broker\n\ndp = Dispatcher()\nbot = Bot(token=\"TOKEN\")\nbot2 = Bot(token=\"TOKEN\")\n\n\n@dp.startup()\nasync def setup_taskiq(bot: Bot, *_args, **_kwargs):\n # Here we check if it's a clien-side,\n # Becuase otherwise you're going to\n # create infinite loop of startup events.\n if not broker.is_worker_process:\n logging.info(\"Setting up taskiq\")\n await broker.startup()\n\n\n@dp.shutdown()\nasync def shutdown_taskiq(bot: Bot, *_args, **_kwargs):\n if not broker.is_worker_process:\n logging.info(\"Shutting down taskiq\")\n await broker.shutdown()\n\n\nasync def main():\n await dp.start_polling(bot, bot2)\n\n\nif __name__ == \"__main__\":\n logging.basicConfig(level=logging.INFO, stream=sys.stdout)\n asyncio.run(main())\n\n```\n\nThe only thing that left is to add few lines to your broker definition.\n\n\n```python\n# Please use real broker for taskiq.\nfrom taskiq_broker import MyBroker\nimport taskiq_aiogram\n\nbroker = MyBroker()\n\n# This line is going to initialize everything.\ntaskiq_aiogram.init(\n broker,\n \"your_project.__main__:dp\",\n \"your_project.__main__:bot\",\n \"your_project.__main__:bot2\",\n)\n```\n\nThat's it.\n\nLet's create some tasks! I created task in a separate module,\nnamed `tasks.py`.\n\n```python\nfrom aiogram import Bot\nfrom your_project.tkq import broker\n\n@broker.task(task_name=\"my_task\")\nasync def my_task(chat_id: int, bot: Bot = TaskiqDepends()) -> None:\n print(\"I'm a task\")\n await asyncio.sleep(4)\n await bot.send_message(chat_id, \"task completed\")\n\n```\n\nNow let's call our new task somewhere in bot commands.\n\n```python\nfrom aiogram import types\nfrom aiogram.filters import Command\n\nfrom tasks import my_task\n\n\n@dp.message(Command(\"task\"))\nasync def message(message: types.Message):\n await my_task.kiq(message.chat.id)\n\n```\n\nTo start the worker, please type:\n\n```\ntaskiq worker your_project.tkq:broker --fs-discover\n```\n\nWe use `--fs-discover` to find all tasks.py modules recursively\nand import all tasks into broker.\n\n\nNow we can fire the task and see everything in action.\n\n\n",
"bugtrack_url": null,
"license": "LICENSE",
"summary": "Taskiq integration with Aiogram",
"version": "0.3.1",
"project_urls": null,
"split_keywords": [
"taskiq",
"tasks",
"distributed",
"async"
],
"urls": [
{
"comment_text": "",
"digests": {
"blake2b_256": "2223bfb229bb6373af141b0d3ddbbd0ebc3a524663a87a9099aa4a492003bead",
"md5": "2578d6be69e2da7ecd60f861ea0e6603",
"sha256": "6332f5cd6db219c8f9e7cee8db5ba2dc96757e12aaddb443394128a9c3264ba6"
},
"downloads": -1,
"filename": "taskiq_aiogram-0.3.1-py3-none-any.whl",
"has_sig": false,
"md5_digest": "2578d6be69e2da7ecd60f861ea0e6603",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.8.1,<4.0.0",
"size": 5176,
"upload_time": "2023-09-10T10:36:26",
"upload_time_iso_8601": "2023-09-10T10:36:26.605390Z",
"url": "https://files.pythonhosted.org/packages/22/23/bfb229bb6373af141b0d3ddbbd0ebc3a524663a87a9099aa4a492003bead/taskiq_aiogram-0.3.1-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": "",
"digests": {
"blake2b_256": "c3d9741ba505b98c1a5fd77996e3d8c1c8d57411352be305dbd8fd5e1b817234",
"md5": "b3fae329123eb89283a05c3df82d4b2d",
"sha256": "0175ac768d7f315318b637eece2ddb24f35a1768990616021c344ca3383186dc"
},
"downloads": -1,
"filename": "taskiq_aiogram-0.3.1.tar.gz",
"has_sig": false,
"md5_digest": "b3fae329123eb89283a05c3df82d4b2d",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.8.1,<4.0.0",
"size": 4575,
"upload_time": "2023-09-10T10:36:27",
"upload_time_iso_8601": "2023-09-10T10:36:27.596207Z",
"url": "https://files.pythonhosted.org/packages/c3/d9/741ba505b98c1a5fd77996e3d8c1c8d57411352be305dbd8fd5e1b817234/taskiq_aiogram-0.3.1.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2023-09-10 10:36:27",
"github": false,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"lcname": "taskiq-aiogram"
}
JSON