from typing import Annotated, Callable, Any from typing_extensions import Doc from aiogram import Bot, Dispatcher from aiogram.client.default import DefaultBotProperties from aiogram.types import Message from aiogram.utils.callback_answer import CallbackAnswerMiddleware from aiogram.utils.i18n import I18n from fastapi import FastAPI from fastapi.applications import Lifespan, AppType from secrets import token_hex from .config import Config from .fsm.db_storage import DbStorage from .middleware.telegram import AuthMiddleware, I18nMiddleware from .model.user import UserBase from .model.entity_metadata import EntityMetadata from .model.descriptors import BotCommand from .router import Router class QBotApp(FastAPI): """ Main class for the QBot application """ def __init__[UserType: UserBase]( self, user_class: ( Annotated[ type[UserType], Doc("User class that will be used in the application") ] | None ) = None, config: Config | None = None, bot_start: ( Annotated[ Callable[ [ Annotated[ Callable[[Message, Any], None], Doc("Default handler for the start command"), ], Message, Any, ], None, ], Doc("Handler for the start command"), ] | None ) = None, lifespan: Lifespan[AppType] | None = None, *args, **kwargs, ): if config is None: config = Config() if user_class is None: from .model.default_user import DefaultUser user_class = DefaultUser self.user_class = user_class self.entity_metadata: EntityMetadata = user_class.entity_metadata self.config = config self.lifespan = lifespan self.bot = Bot( token=self.config.TELEGRAM_BOT_TOKEN, default=DefaultBotProperties(parse_mode="HTML"), ) dp = Dispatcher(storage=DbStorage()) i18n = I18n(path="locales", default_locale="en", domain="messages") i18n_middleware = I18nMiddleware(user_class=user_class, i18n=i18n) i18n_middleware.setup(dp) dp.callback_query.middleware(CallbackAnswerMiddleware()) from .bot.handlers.start import router as start_router dp.include_router(start_router) from .bot.handlers.menu.main import router as main_menu_router auth = AuthMiddleware(user_class=user_class) main_menu_router.message.middleware.register(auth) main_menu_router.callback_query.middleware.register(auth) dp.include_router(main_menu_router) self.dp = dp self.bot_auth_token = token_hex(128) self.start_handler = bot_start self.bot_commands = dict[str, BotCommand]() from .lifespan import default_lifespan super().__init__(lifespan=default_lifespan, *args, **kwargs) from .api_route.telegram import router as telegram_router self.include_router(telegram_router, prefix="/api/telegram", tags=["telegram"]) self.root_router = Router() self.root_router._commands = self.bot_commands self.command = self.root_router.command def register_routers(self, *routers: Router): for router in routers: for command_name, command in router._commands.items(): self.bot_commands[command_name] = command