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.

importasyncioimportloggingimportsysfromaiogramimportBot, Dispatcher# Your taskiq brokerfromyour_project.tkqimportbrokerdp=Dispatcher()
bot=Bot(token="TOKEN")
bot2=Bot(token="TOKEN")
@dp.startup()asyncdefsetup_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.ifnotbroker.is_worker_process:
logging.info("Setting up taskiq")
awaitbroker.startup()
@dp.shutdown()asyncdefshutdown_taskiq(bot: Bot, *_args, **_kwargs):
ifnotbroker.is_worker_process:
logging.info("Shutting down taskiq")
awaitbroker.shutdown()
asyncdefmain():
awaitdp.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.

# Please use real broker for taskiq.fromtaskiq_brokerimportMyBrokerimporttaskiq_aiogrambroker=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.

fromaiogramimportBotfromyour_project.tkqimportbroker@broker.task(task_name="my_task")asyncdefmy_task(chat_id: int, bot: Bot=TaskiqDepends()) ->None:
print("I'm a task")
awaitasyncio.sleep(4)
awaitbot.send_message(chat_id, "task completed")

Now let's call our new task somewhere in bot commands.

fromaiogramimporttypesfromaiogram.filtersimportCommandfromtasksimportmy_task@dp.message(Command("task"))asyncdefmessage(message: types.Message):
awaitmy_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.