megabot 0.0.6

A library for simultaneous operation of a large number of telegram bots with a single code base.

Python - telegram API adapter. Based in Pydantic model libraries

You need to implement a web interface for a webhook yourself. I recommend FastAPI.

pip install megabot

1. Fill in the dispatcher tokens in a convenient way for you in the format dict(int(bot_id), str(token)
2. Define callback, message, and command handlers
3. Connect the dispatcher to processing incoming webhooks.
4. Done!

The filtering of the callbacks was taken from aiogram, the syntax is identical

example main.py

from megabot.dispatcher import Dispatcher
from megabot.poster import MessageService
from megabot.utils.callback_data import CallbackData

menu_cat = CallbackData('cat', 'id') # Callback fabric 
message_service = MessageService()

async def hello_word(callback: CallbackQuery):
    print(callback.data)
    await message_service.callback_answer(self.dp.tokens[bot_id], callback.id)
    
dp = Dispatcher()

bot_id = 9***9
token = '9***9:AA*****AA'
dp.tokens[bot_id] = token

dp.register_handler_callback(hello_word, callback_filter=menu_cat.filter())

webhook router file:

from main import dp

async def data_router(self, bot_id: int, data: dict) -> NoReturn:
    if not(dp.tokens.get(bot_id, None)):
        dp.tokens[bot_id] = token

    await dp.message_router(bot_id, data)

for local testing: NOT FOR PROD!!! add to main.py

from megabot.poster import get_updates
offset = 0

async def check_updates():
    global offset
    while True:
        result, error = await get_updates(token, offset)
        if result:
            result_list = result['result']
            if len(result_list) > 0:
                print(result)
                offset = result_list[-1]['update_id']+1
            await dp.message_router(bot_id, result)
        else:
            print(error)
        await asyncio.sleep(1)


await def start_bot()
    asyncio.create_task(check_updates())

add start_bot() to main file web_hooks