""" Dependency Injection Container для управления зависимостями Следует принципам SOLID, особенно Dependency Inversion Principle """ from typing import Dict, Any, TypeVar, Type from sqlalchemy.ext.asyncio import AsyncSession from src.interfaces.base import ( IUserRepository, ILotteryRepository, IParticipationRepository, IWinnerRepository, ILotteryService, IUserService, IBotController, IKeyboardBuilder, IMessageFormatter ) from src.repositories.implementations import ( UserRepository, LotteryRepository, ParticipationRepository, WinnerRepository ) from src.components.services import LotteryServiceImpl, UserServiceImpl from src.components.ui import KeyboardBuilderImpl, MessageFormatterImpl from src.controllers.bot_controller import BotController T = TypeVar('T') class DIContainer: """Контейнер для dependency injection""" def __init__(self): self._services: Dict[Type, Any] = {} self._singletons: Dict[Type, Any] = {} # Регистрируем singleton сервисы self.register_singleton(IKeyboardBuilder, KeyboardBuilderImpl) self.register_singleton(IMessageFormatter, MessageFormatterImpl) def register_singleton(self, interface: Type[T], implementation: Type[T]): """Зарегистрировать singleton сервис""" self._services[interface] = implementation def register_transient(self, interface: Type[T], implementation: Type[T]): """Зарегистрировать transient сервис""" self._services[interface] = implementation def get_singleton(self, interface: Type[T]) -> T: """Получить singleton экземпляр""" if interface in self._singletons: return self._singletons[interface] if interface not in self._services: raise ValueError(f"Service {interface} not registered") implementation = self._services[interface] instance = implementation() self._singletons[interface] = instance return instance def create_scoped_container(self, session: AsyncSession) -> 'ScopedContainer': """Создать scoped контейнер для сессии базы данных""" return ScopedContainer(self, session) class ScopedContainer: """Scoped контейнер для одной сессии базы данных""" def __init__(self, parent: DIContainer, session: AsyncSession): self.parent = parent self.session = session self._instances: Dict[Type, Any] = {} def get(self, interface: Type[T]) -> T: """Получить экземпляр сервиса""" # Если это singleton, получаем из родительского контейнера if interface in [IKeyboardBuilder, IMessageFormatter]: return self.parent.get_singleton(interface) # Если уже создан в текущем scope, возвращаем if interface in self._instances: return self._instances[interface] # Создаем новый экземпляр instance = self._create_instance(interface) self._instances[interface] = instance return instance def _create_instance(self, interface: Type[T]) -> T: """Создать экземпляр с разрешением зависимостей""" if interface == IUserRepository: return UserRepository(self.session) elif interface == ILotteryRepository: return LotteryRepository(self.session) elif interface == IParticipationRepository: return ParticipationRepository(self.session) elif interface == IWinnerRepository: return WinnerRepository(self.session) elif interface == ILotteryService: return LotteryServiceImpl( self.get(ILotteryRepository), self.get(IParticipationRepository), self.get(IWinnerRepository), self.get(IUserRepository) ) elif interface == IUserService: return UserServiceImpl( self.get(IUserRepository) ) elif interface == IBotController: return BotController( self.get(ILotteryService), self.get(IUserService), self.get(IKeyboardBuilder), self.get(IMessageFormatter), self.get(ILotteryRepository), self.get(IParticipationRepository) ) else: raise ValueError(f"Cannot create instance of {interface}") # Глобальный экземпляр контейнера container = DIContainer()