refactor(middlewares): message_info getter mixin

src/tg_bot/middlewares/error_handler.py

@@ -5,8 +5,10 @@
 from aiogram.types import CallbackQuery, Message, TelegramObject
 from errors import BaseMsgError
 
+from ..mixins import MessageInfo, MessageInfoGetterMixin
 
-class ErrorHandlerMiddleware(BaseMiddleware):
+
+class ErrorHandlerMiddleware(BaseMiddleware, MessageInfoGetterMixin):
     def __init__(self) -> None:
         self.logger = structlog.get_logger("error_handler_middleware")
 
@@ -17,18 +19,27 @@ async def __call__(
         data: Dict[str, Any],
     ) -> Any:
         try:
-            await handler(event, data)
-        except BaseMsgError as e:
-            if isinstance(event, CallbackQuery):
-                self.logger.error(e.msg, username=event.from_user.username, chat_id=event.from_user.id)
-                await event.answer(e.msg)
-            if isinstance(event, Message):
-                self.logger.error(e.msg, username=event.chat.username, chat_id=event.chat.id)
-                await event.answer(e.msg)
+            return await handler(event, data)
         except Exception as e:
-            if isinstance(event, CallbackQuery):
-                self.logger.error(e, username=event.from_user.username, chat_id=event.from_user.id, exc_info=True)
-                await event.answer("Unexpected error")
-            if isinstance(event, Message):
-                self.logger.error(e, username=event.chat.username, chat_id=event.chat.id, exc_info=True)
-                await event.answer("Unexpected error")
+            error = e
+        message_info = self._get_message_info(event)
+        if isinstance(error, BaseMsgError):
+            await self._process_msg_error(error, event, message_info)
+        else:
+            await self._process_unexpected_error(error, event, message_info)
+
+    async def _process_msg_error(self, error: BaseMsgError, event: TelegramObject, message_info: MessageInfo):
+        if not message_info.user:
+            self.logger.error("No user info", error=error.msg)
+            return
+        self.logger.error(error.msg, username=message_info.user.username, chat_id=message_info.user.id)
+        if isinstance(event, CallbackQuery | Message):
+            await event.answer(error.msg)
+
+    async def _process_unexpected_error(self, error: Exception, event: TelegramObject, message_info: MessageInfo):
+        if not message_info.user:
+            self.logger.error("No user info", error=error, exc_info=True)
+            return
+        self.logger.error(error, username=message_info.user.username, chat_id=message_info.user.id, exc_info=True)
+        if isinstance(event, CallbackQuery | Message):
+            await event.answer("Unexpected error")

src/tg_bot/middlewares/is_admin.py

@@ -4,8 +4,10 @@
 from aiogram import BaseMiddleware
 from aiogram.types import TelegramObject
 
+from ..mixins import MessageInfoGetterMixin
 
-class BlockNonAdminMiddleware(BaseMiddleware):
+
+class BlockNonAdminMiddleware(BaseMiddleware, MessageInfoGetterMixin):
     def __init__(self, admin_id: int) -> None:
         self.admin_id = admin_id
         self.logger = structlog.get_logger("blocknonadmin_middleware")
@@ -24,5 +26,9 @@ async def __call__(
         event: TelegramObject,
         data: Dict[str, Any],
     ) -> Any:
-        if self._is_admin(event.chat.id):  # type: ignore
+        message_info = self._get_message_info(event)
+        if message_info.user is None:
+            return await handler(event, data)
+        if self._is_admin(message_info.user.id):
             return await handler(event, data)
+        self.logger.warning("Block non-admin user", user_id=message_info.user.id, username=message_info.user.username)

src/tg_bot/middlewares/logging.py

@@ -4,8 +4,10 @@
 from aiogram import BaseMiddleware
 from aiogram.types import TelegramObject
 
+from ..mixins import MessageInfoGetterMixin
 
-class LogCommandsMiddleware(BaseMiddleware):
+
+class LogCommandsMiddleware(BaseMiddleware, MessageInfoGetterMixin):
     def __init__(self) -> None:
         self.logger = structlog.get_logger("logger_middleware")
 
@@ -15,5 +17,8 @@ async def __call__(
         event: TelegramObject,
         data: Dict[str, Any],
     ) -> Any:
-        self.logger.info(event.text, username=event.chat.username, chat_id=event.chat.id)  # type: ignore
+        message_info = self._get_message_info(event)
+        if message_info.user is None:
+            return await handler(event, data)
+        self.logger.info(message_info.text, username=message_info.user.username, chat_id=message_info.user.id)
         return await handler(event, data)

src/tg_bot/mixins.py

@@ -0,0 +1,24 @@
+from dataclasses import dataclass
+
+from aiogram.types import CallbackQuery, Message, TelegramObject, User
+
+
+@dataclass
+class MessageInfo:
+    user: User | None
+    text: str | None
+
+
+class MessageInfoGetterMixin:
+    @staticmethod
+    def _get_message_info(event: TelegramObject) -> MessageInfo:
+        event_text = None
+        user = None
+
+        if isinstance(event, Message):
+            user = event.from_user if event.from_user else None
+            event_text = event.text
+        if isinstance(event, CallbackQuery) and not user:
+            user = event.from_user if event.from_user else None
+
+        return MessageInfo(user=user, text=event_text)