Andrey
e2b1353408
- Создана система отслеживания миграций (MigrationRepository, таблица migrations) - Добавлен скрипт apply_migrations.py для автоматического применения миграций - Созданы CI/CD пайплайны (.github/workflows/ci.yml, deploy.yml) - Обновлена документация по миграциям в database-patterns.md - Миграции применяются автоматически при деплое в продакшн
73 lines
3.0 KiB
Python
73 lines
3.0 KiB
Python
from typing import Any, Dict
|
||
|
||
try:
|
||
from typing import Annotated
|
||
except ImportError:
|
||
from typing_extensions import Annotated
|
||
|
||
from aiogram import BaseMiddleware
|
||
from aiogram.types import TelegramObject
|
||
from helper_bot.utils.base_dependency_factory import get_global_instance
|
||
from helper_bot.utils.helper_func import check_access
|
||
from logs.custom_logger import logger
|
||
|
||
|
||
class AdminAccessMiddleware(BaseMiddleware):
|
||
"""Middleware для проверки административного доступа"""
|
||
|
||
async def __call__(self, handler, event: TelegramObject, data: Dict[str, Any]) -> Any:
|
||
if hasattr(event, 'from_user'):
|
||
user_id = event.from_user.id
|
||
username = getattr(event.from_user, 'username', 'Unknown')
|
||
|
||
logger.info(f"AdminAccessMiddleware: проверка доступа для пользователя {username} (ID: {user_id})")
|
||
|
||
# Получаем bot_db из data (внедренного DependenciesMiddleware)
|
||
bot_db = data.get('bot_db')
|
||
if not bot_db:
|
||
# Fallback: получаем напрямую если middleware не сработала
|
||
bdf = get_global_instance()
|
||
bot_db = bdf.get_db()
|
||
|
||
is_admin_result = await check_access(user_id, bot_db)
|
||
logger.info(f"AdminAccessMiddleware: результат проверки для {username}: {is_admin_result}")
|
||
|
||
if not is_admin_result:
|
||
logger.warning(f"AdminAccessMiddleware: доступ запрещен для пользователя {username} (ID: {user_id})")
|
||
if hasattr(event, 'answer'):
|
||
await event.answer('Доступ запрещен!')
|
||
return
|
||
|
||
try:
|
||
# Вызываем хендлер с data
|
||
return await handler(event, data)
|
||
except TypeError as e:
|
||
if "missing 1 required positional argument: 'data'" in str(e):
|
||
logger.error(f"Ошибка в AdminAccessMiddleware: {e}. Хендлер не принимает параметр 'data'")
|
||
# Пытаемся вызвать хендлер без data (для совместимости с MagicData)
|
||
return await handler(event)
|
||
else:
|
||
logger.error(f"TypeError в AdminAccessMiddleware: {e}")
|
||
raise
|
||
except Exception as e:
|
||
logger.error(f"Неожиданная ошибка в AdminAccessMiddleware: {e}")
|
||
raise
|
||
|
||
|
||
# Dependency providers
|
||
def get_bot_db():
|
||
"""Провайдер для получения экземпляра БД"""
|
||
bdf = get_global_instance()
|
||
return bdf.get_db()
|
||
|
||
|
||
def get_settings():
|
||
"""Провайдер для получения настроек"""
|
||
bdf = get_global_instance()
|
||
return bdf.settings
|
||
|
||
|
||
# Type aliases for dependency injection
|
||
BotDB = Annotated[object, get_bot_db()]
|
||
Settings = Annotated[dict, get_settings()]
|