如何将用户的 id 作为参数传递到装饰器中。爱图

Question

我正在编写一个简单的程序机器人，其中一项功能需要授权。我发现的最简单的方法是拥有一组电报用户 ID，当用户想要执行命令 /log 时，它会检查他的 ID 是否在“管理员”列表中。所以我最终得到了这个：

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')

但问题是

@is_authorized(message.from_user.id)

“message”在使用前没有被引用。 如何将“message.from_user.id”作为参数传递给我的 is_authorized 装饰器？ 我不太喜欢创建类对象或维护会话，我想保持简单

Answer 1

我们不能通过直接在 aiogram 中创建装饰器来影响处理程序。

aiogram 提供了强大的机制，可以通过 中间件自定义事件处理程序。

根据aiogram文档，中间件的工作原理如下

在这种情况下，我们需要内部中间件（Middleware），因为我们的函数需要在过滤器之后运行

Command(commands=["log"])

Aiogram3

有一个标记处理程序，可以在此中间件中使用，称为 flags。适合你的情况。

首先我们需要创建中间件。

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())