pythontelegramaiogram

How to pass user's id into the decorator as an argument. aiogram


Im writing a simple procedure bot and one of the features requires authorization. Easiest way I found is having an array of telegram user ids and when user wants to execute a command /log it would check if his id is in "admins" list. So I ended up having this:

admins = [123123, 12312]
def is_authorized(user_id):
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            if user_id in users:
                return func(*args, **kwargs)
            else:
                return
        return wrapper

    return decorator


@dp.message(Command(commands=["log"]))
@is_authorized(message.from_user.id)
async def loglast20(message: types.Message):
    await bot.send_message(message.from_user.id, 'some logs here')

But the problem is

@is_authorized(message.from_user.id)

'message' is not referenced before usage. How do I pass 'message.from_user.id' as an argument to my is_authorized decorator? Im not really into creating class objects or maintaining sessions, i'd like to keep it simple


Solution

  • We cannot affect the handlers by creating decorators directly in the aiogram.

    aiogram provides powerful mechanism for customizing event handlers via middlewares.

    According to the aiogram documentation, the middleware works as follows How to work middleware

    In this case we need inner middleware (Middleware) because our function needs to run after the filter Command(commands=["log"])

    Aiogram3 has a marker handler that can be used in this middleware called flags. It is suitable for your situation.

    First we need to create middleware.

    middleware.py

    import logging
    from typing import Any, Awaitable, Callable, Dict
    
    from aiogram import BaseMiddleware
    from aiogram.dispatcher.flags import get_flag
    from aiogram.types import Message, TelegramObject
    
    logger = logging.getLogger(__name__)
    
    ADMINS = [123123, 12312]
    
    
    class AuthorizationMiddleware(BaseMiddleware):
        """
        Helps to check if user is authorized to use the bot
        """
    
        async def __call__(
                self,
                handler: Callable[[TelegramObject, Dict[str, Any]], Awaitable[Any]],
                event: TelegramObject,
                data: Dict[str, Any],
        ) -> Any:
            if not isinstance(event, Message):
                return await handler(event, data)
    
            authorization = get_flag(data, "authorization")
            print("Authorization: ", authorization)
            if authorization is not None:
                if authorization["is_authorized"]:
                    if event.chat.id in ADMINS:
                        return await handler(event, data)
                    else:
                        return None
            else:
                return await handler(event, data)
    

    bot.py

    import asyncio
    import logging
    import sys
    
    from aiogram import Bot, Dispatcher, types, flags
    from aiogram.enums import ParseMode
    from aiogram.filters import CommandStart, Command
    from aiogram.types import Message, CallbackQuery
    from aiogram.utils.markdown import hbold
    
    
    from middleware import AuthorizationMiddleware
    
    BOT_TOKEN = "Your bot token"
    
    dp = Dispatcher()
    
    
    @dp.message(CommandStart())
    async def command_start_handler(message: Message) -> None:
        await message.answer(f"Hello, {hbold(message.from_user.full_name)}!")
    
    
    @dp.message(Command(commands=["log"]))
    @flags.authorization(is_authorized=True) # You need to write this @flags line after the @dp.message()
    async def log_last_20(message: types.Message, bot: Bot):
        await bot.send_message(message.from_user.id, 'some logs here')
    
    
    
    async def main() -> None:
        bot = Bot(BOT_TOKEN, parse_mode=ParseMode.HTML)
        dp.message.middleware(AuthorizationMiddleware())
        await dp.start_polling(bot)
    
    
    if __name__ == "__main__":
        logging.basicConfig(level=logging.INFO, stream=sys.stdout)
        asyncio.run(main())