Фильтрация по типу пользователя (передача данных с промежуточного слоя диспетчера на фильтры роутеров в iogram
Суть проблемы:
Пишу корпоративного бота для обработки обращений подчиненных по возникающим рабочим моментам. Одна из задач заключается в разделении типов пользователей по типам сессий (отдельных роутеры по целевому назначению для определенного состава людей). В локальной базе данных проекта имеются данные о пользователях и принадлежность к типу сессии (чтобы была возможность разграничивать доступ).
Логика: при отправке любых сообщений в outer_middleware идет запрос к базе данных:
Код в main.py: dp.message.outer_middleware(TypeSessionMiddleware(session_pool=session_pool_LOCAL_DB)).
Код в модуле с промежуточными слоями для TypeSessionMiddleware:
class TypeSessionMiddleware(BaseMiddleware):
"""
Определитель пользователей.
На основе from_user.id выдает текстовый тип сессии session_types.
Далее эти значения передаются на роутер и там сравниваются в кастомных фильтрах.
Таким образом, достигается фильтрация пользователей по сессиям (разграничение прав).
"""
def __init__(self, session_pool: async_sessionmaker) -> Any:
self.session_pool = session_pool
async def __call__(
self,
handler: Callable[[TelegramObject, Dict[str, Any]], Awaitable[Any]],
event: TelegramObject,
data: Dict[str, Any],
) -> Any:
# Проверка принадлежности сообщения:
if isinstance(event, Message):
get_id_tg = event.from_user.id
query = select(Users.session_type).where(Users.id_tg == get_id_tg)
async with self.session_pool() as session:
get_session_types = await session.execute(query)
session_type_str = get_session_types.scalar_one_or_none()
# print(session_type_str)
await session.commit()
# Передаем в словарик данных наш тип сессии:
data['session_type'] = session_type_str
return await handler(event, data)
В этом куске кода проблем нет, все нормально отрабатывает.
Далее проблема возникает с кастомным фильтром для роутеров:
class TypeSessionFilter(Filter):
def __init__(self, allowed_types: list[str]):
self.allowed_types = allowed_types # <- session_type_str разрешенные типы
async def __call__(self, message: Message, data: Dict[str, Any] ) -> bool: # event: TelegramObject
get_session_type_in_data = data.get("session_type")
# get_session_type_in_data = data["session_type"]
print(f'Итог: {get_session_type_in_data}')
return get_session_type_in_data in self.allowed_types
Ошибка:
TypeError: TypeSessionFilter.call() missing 1 required positional argument: 'data'
Что бы я не делал, не могу получить доступ в фильтре к словарику данных, что по задумке создателей aiogram должен перебрасываться между всеми обработчиками и фильтрами (на сколько я это понял).
Вот инициализация фильтра на роутере в модуле для одного из типа сессий
retail_router.message.filter(ChatTypeFilter(['private']), TypeSessionFilter(allowed_types=['oait']))
Других решений не нашел, "ChatGPT4" выдает мусор. Что я делаю не так?
Ответы (1 шт):
Я решил проблему сам. Возможно это костыль, но другим образом мне не удалось решить проблему. Частично, решение подсмотрел здесь (
youtu.be/55w2QpPGC-E?si=-949tq0Qe63h1QI4
), минута - 6,55.
Вот изменения в TypeSessionMiddleware(): 1) добавляем self.session_type_str = None в init. 2) получаем bot: Bot = data.get('bot') - (важная пометка: в родительском классе Bot добавляем переменные self.get_type_session_for_admin = None.
Код промежуточного слоя для диспетчера:
class TypeSessionMiddleware(BaseMiddleware):
"""
Универсальный слой-определитель пользователей.
На основе from_user.id выдает текстовый тип сессии \
(session_types = ['admin', 'retail', '*', '*', '*', '*', '*'])
Далее эти значения передаются на роутер и там сравниваются в кастомных \
фильтрах.
Таким образом, достигается фильтрация пользователей по сессиям \
(разграничение прав).
"""
def __init__(self, session_pool: async_sessionmaker) -> Any:
self.session_pool = session_pool
self.session_type_str = None
async def __call__(
self,
handler: Callable[[TelegramObject, Dict[str, Any]], Awaitable[Any]],
event: TelegramObject,
data: Dict[str, Any]
) -> Any:
# self.data = data
bot: Bot = data.get('bot')
# Проверка принадлежности сообщения:
if isinstance(event, Message):
get_id_tg = event.from_user.id
query = select(Users.session_type).where(Users.id_tg == get_id_tg)
async with self.session_pool() as session:
get_session_types = await session.execute(query)
self.session_type_str = get_session_types.scalar_one_or_none()
await session.commit()
# Передаем в словарик данных наш тип сессии:
(data["session_type"] = self.session_pool.get_session_type(event))
# Если тип сессии из базы (разрешенный) совпадает \
# со значением в фильтре:
bot.get_type_session = self.session_type_str # Присваиваем значение \
# переменной в корневом классе Bot.
print(f'Наш тип сесcии: {bot.get_type_session}')
return await handler(event, data)
Логика такова: далее по коду мы увидим, что из промежуточного слоя данные из бд передаются в эти переменные (самописные переменные класса Bot). Однако, пока не ясно еще, как эта конструкция будет отрабатывать, при одновременном срабатывании на разных пользователей).
Теперь наш тип сессии определяется один раз в dp.outer_middleware, а далее передается в роутеры на фильтры.
Как реализован фильтр:
class TypeSessionFilter(Filter):
def __init__(self, allowed_types: list[str]):
self.allowed_types = allowed_types # <- session_type_str разрешенные типы
async def __call__(self, message: Message) -> bool:
bot = message.bot
# print(f'Пришло в фильтр: {bot.get_type_session}')
return bot.get_type_session in self.allowed_types
Теперь фильтр отрабатывает отлично. Единственное, пока не тестировал на одновременное срабатывание различных типов сессий. Вот полный код: https://github.com/ArtemXYZ/bot_dns_reboot . Если есть более изящное решение, подскажите.