Оптимизация и улучшение middleware и обработки сессий базы данных

- Рефакторинг middlewares с добавлением типизации и улучшенной логики
- Оптимизация SessionMiddleware с использованием пула соединений
- Улучшение обработки пользователей и логирования в UserMiddleware
- Расширение функциональности ThrottlingMiddleware с более гибким управлением
- Добавление дополнительных проверок и логирования в AdminMiddleware
- Оптимизация функции upsert_user с поддержкой передачи существующей сессии
This commit is contained in:
izzzzzi
2025-02-25 02:39:44 +05:00
parent e3a9ff4dfd
commit b08040f46b
7 changed files with 225 additions and 83 deletions
+39 -8
View File
@@ -1,27 +1,58 @@
from collections.abc import Awaitable, Callable
from typing import Any
from typing import Any, Dict, Set, cast
from aiogram import BaseMiddleware
from aiogram.types import TelegramObject
from aiogram.types import Message, CallbackQuery, TelegramObject
from config import ADMIN_ID
class AdminMiddleware(BaseMiddleware):
"""Middleware для проверки прав администратора.
Добавляет в data['admin'] = True/False в зависимости от того,
является ли пользователь администратором.
"""
# Кэш ID администраторов для быстрого доступа
_admin_ids: Set[int] = set(ADMIN_ID) if isinstance(ADMIN_ID, (list, tuple)) else {ADMIN_ID}
async def __call__(
self,
handler: Callable[[TelegramObject, dict[str, Any]], Awaitable[Any]],
handler: Callable[[TelegramObject, Dict[str, Any]], Awaitable[Any]],
event: TelegramObject,
data: dict[str, Any],
data: Dict[str, Any],
) -> Any:
"""Обрабатывает событие и добавляет флаг администратора в data.
Args:
handler: Обработчик события
event: Событие Telegram
data: Словарь с данными события
Returns:
Результат выполнения обработчика
"""
data["admin"] = self._check_admin_access(event)
return await handler(event, data)
def _check_admin_access(self, event: TelegramObject) -> bool:
"""Проверяет, имеет ли пользователь права администратора.
Args:
event: Событие Telegram
Returns:
True, если пользователь администратор, иначе False
"""
try:
admin_ids: int | list[int] = ADMIN_ID
if isinstance(admin_ids, list):
return event.from_user.id in admin_ids
return event.from_user.id == admin_ids
if isinstance(event, Message):
return event.from_user and event.from_user.id in self._admin_ids
elif isinstance(event, CallbackQuery):
return event.from_user and event.from_user.id in self._admin_ids
# Для других типов событий пытаемся получить from_user
user_id = getattr(getattr(event, 'from_user', None), 'id', None)
return user_id in self._admin_ids if user_id else False
except Exception:
return False