Проблема с получением "data" (Aiogram 3.20, Middleware, RoleFilter)
Проблема с получением "data" как аргумент. Создал модуль с такой же задачей. В данной итерации ошибку не получаю, но точно вижу из логов что "роль" не передается. Подскажите кто знает, куда копать ?
Лог из консоли:
Middleware: user_id=XXXXXXXXX, role=admin - команда /start
Filter: user_role=None, required_role=admin, result=False - команда /admin
Код:
DB_URL = "sqlite+aiosqlite:///./test3.db"
engine = create_async_engine(DB_URL, echo=False)
session_maker = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
bot = Bot(token=BOT_TOKEN)
Base = declarative_base()
class RoleEnum(Enum):
ADMIN = "admin"
USER = "user"
class User(Base):
__tablename__ = "users"
user_id = Column(Integer, primary_key=True)
role = Column(SAEnum(RoleEnum), nullable=False)
# --- Middleware ---
class RoleMiddleware(BaseMiddleware):
def __init__(self, session_pool: async_sessionmaker):
self.session_pool = session_pool
super().__init__()
async def __call__(
self,
handler: Callable[[TelegramObject, Dict[str, Any]], Awaitable[Any]],
event: types.Message,
data: Dict[str, Any]
) -> Any:
user_id = event.from_user.id
async with self.session_pool() as session:
user = await session.get(User, user_id)
if not user:
user = User(user_id=user_id, role=RoleEnum.USER)
session.add(user)
await session.commit()
# Передаем роль через data
data["user_role"] = user.role.value
print(f"Middleware: user_id={user_id}, role={user.role.value}")
return await handler(event, data)
# --- Фильтр ---
class RoleFilter(Filter):
def __init__(self, role: RoleEnum):
self.role = role.value
async def __call__(self, event: types.Message, **kwargs) -> bool:
data = kwargs.get("data", {})
user_role = data.get("user_role") # Извлекаем роль из data
result = user_role == self.role
print(f"Filter: user_role={user_role}, required_role={self.role}, result={result}")
return result
# --- Обработчики ---
async def start_handler(message: types.Message, **kwargs):
data = kwargs.get("data", {})
await message.answer(f"Ваша роль: {data.get('user_role', 'неизвестно')}")
async def admin_handler(message: types.Message, data: Dict[str, Any]):
await message.answer("Привет, админ!")
# --- Функция для изменения роли пользователя ---
async def set_admin_role(user_id: int):
async with session_maker() as session:
user = await session.get(User, user_id)
if user:
user.role = RoleEnum.ADMIN
await session.commit()
else:
print(f"Пользователь с user_id={user_id} не найден.")
# --- Основной код ---
async def main():
# Создаем таблицы
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
dp = Dispatcher()
# Подключаем middleware
dp.message.middleware(RoleMiddleware(session_pool=session_maker))
dp.callback_query.middleware(RoleMiddleware(session_pool=session_maker))
# Регистрируем обработчики
dp.message.register(start_handler, Command("start"))
dp.message.register(admin_handler, RoleFilter(RoleEnum.ADMIN))
# Изменения роли
await set_admin_role(XXXXXXXX)
# Запуск бота
await dp.start_polling(bot)