Приведи пример применения принципа Dependency Inversion

Пример применения принципа Dependency Inversion

Dependency Inversion Principle (DIP) — это последний принцип SOLID. Он утверждает, что высокоуровневые модули не должны зависеть от низкоуровневых модулей. Обе должны зависеть от абстракций.

Суть принципа

❌ НЕПРАВИЛЬНО (Tight Coupling):
HighLevelModule → ConcreteImplementation

✅ ПРАВИЛЬНО (Loose Coupling):
HighLevelModule → Abstraction ← ConcreteImplementation

Пример 1: Email рассылка

# ❌ БЕЗ Dependency Inversion
# Класс зависит от конкретной реализации EmailService

class EmailService:
    def send(self, email: str, message: str) -> bool:
        print(f"Отправка на {email}: {message}")
        return True

class OrderProcessor:
    def __init__(self):
        self.email_service = EmailService()  # ❌ Жёсткая зависимость!
    
    def process_order(self, order_id: int, customer_email: str):
        # Обработка заказа
        order = self.get_order(order_id)
        
        # Отправить уведомление
        self.email_service.send(
            customer_email,
            f"Заказ {order_id} обработан"
        )

# Проблемы:
# 1. Нельзя тестировать без реального EmailService
# 2. Нельзя заменить на SMS или Slack уведомление
# 3. Если EmailService упадёт, упадёт и OrderProcessor

# ✅ С Dependency Inversion
# OrderProcessor зависит от абстракции NotificationService

from abc import ABC, abstractmethod
from typing import Protocol

# Определяем контракт (абстракцию)
class NotificationService(ABC):
    @abstractmethod
    def send(self, recipient: str, message: str) -> bool:
        pass

# Конкретные реализации
class EmailService(NotificationService):
    def send(self, email: str, message: str) -> bool:
        print(f"📧 Email отправлен на {email}: {message}")
        return True

class SMSService(NotificationService):
    def send(self, phone: str, message: str) -> bool:
        print(f"📱 SMS отправлено на {phone}: {message}")
        return True

class SlackService(NotificationService):
    def send(self, channel: str, message: str) -> bool:
        print(f"💬 Slack сообщение в {channel}: {message}")
        return True

# Высокоуровневый модуль зависит от абстракции
class OrderProcessor:
    def __init__(self, notification_service: NotificationService):
        # ✅ Инъекция зависимости через конструктор
        self.notification_service = notification_service
    
    def process_order(self, order_id: int, recipient: str):
        # Обработка заказа
        order = self.get_order(order_id)
        
        # Используем абстракцию, не конкретную реализацию
        self.notification_service.send(
            recipient,
            f"Заказ {order_id} обработан"
        )
    
    def get_order(self, order_id: int):
        return {"id": order_id, "status": "processed"}

# Использование
print("=== С Email ===")
email_service = EmailService()
processor = OrderProcessor(email_service)
processor.process_order(123, "user@example.com")

print("\n=== С SMS ===")
sms_service = SMSService()
processor = OrderProcessor(sms_service)
processor.process_order(123, "+79991234567")

print("\n=== Со Slack ===")
slack_service = SlackService()
processor = OrderProcessor(slack_service)
processor.process_order(123, "#notifications")

Пример 2: Работа с БД

# ❌ БЕЗ DIP: UserService зависит от конкретной БД

class MySQLDatabase:
    def query(self, sql: str):
        # Реальное подключение к MySQL
        pass

class UserService:
    def __init__(self):
        self.db = MySQLDatabase()  # ❌ Жёсткая связь!
    
    def get_user(self, user_id: int):
        return self.db.query(f"SELECT * FROM users WHERE id = {user_id}")

# Если нужно перейти на PostgreSQL или MongoDB — нужно переписывать UserService!

# ✅ С DIP: UserService зависит от абстракции

from abc import ABC, abstractmethod
from typing import Optional, Dict, Any

# Абстракция
class Database(ABC):
    @abstractmethod
    def query(self, sql: str) -> Any:
        pass
    
    @abstractmethod
    def insert(self, sql: str, params: Dict) -> int:
        pass
    
    @abstractmethod
    def update(self, sql: str, params: Dict) -> int:
        pass

# Конкретные реализации
class MySQLDatabase(Database):
    def __init__(self, host: str, user: str, password: str):
        self.host = host
        self.user = user
        self.password = password
    
    def query(self, sql: str) -> Any:
        # Подключение к MySQL
        print(f"[MySQL] Выполняю: {sql}")
        return {"id": 1, "name": "John"}
    
    def insert(self, sql: str, params: Dict) -> int:
        print(f"[MySQL] INSERT: {sql} с {params}")
        return 1
    
    def update(self, sql: str, params: Dict) -> int:
        print(f"[MySQL] UPDATE: {sql} с {params}")
        return 1

class PostgreSQLDatabase(Database):
    def __init__(self, host: str, user: str, password: str):
        self.host = host
        self.user = user
        self.password = password
    
    def query(self, sql: str) -> Any:
        print(f"[PostgreSQL] Выполняю: {sql}")
        return {"id": 1, "name": "John"}
    
    def insert(self, sql: str, params: Dict) -> int:
        print(f"[PostgreSQL] INSERT: {sql} с {params}")
        return 1
    
    def update(self, sql: str, params: Dict) -> int:
        print(f"[PostgreSQL] UPDATE: {sql} с {params}")
        return 1

class MongoDBDatabase(Database):
    def __init__(self, connection_string: str):
        self.connection_string = connection_string
    
    def query(self, sql: str) -> Any:
        print(f"[MongoDB] Выполняю: {sql}")
        return {"_id": "1", "name": "John"}
    
    def insert(self, sql: str, params: Dict) -> int:
        print(f"[MongoDB] INSERT: {sql} с {params}")
        return 1
    
    def update(self, sql: str, params: Dict) -> int:
        print(f"[MongoDB] UPDATE: {sql} с {params}")
        return 1

# Высокоуровневый модуль зависит от абстракции
class UserService:
    def __init__(self, db: Database):  # ✅ Инъекция абстракции
        self.db = db
    
    def get_user(self, user_id: int) -> Optional[Dict]:
        return self.db.query(f"SELECT * FROM users WHERE id = {user_id}")
    
    def create_user(self, name: str, email: str) -> int:
        return self.db.insert(
            "INSERT INTO users (name, email) VALUES (?, ?)",
            {"name": name, "email": email}
        )
    
    def update_user(self, user_id: int, name: str) -> int:
        return self.db.update(
            "UPDATE users SET name = ? WHERE id = ?",
            {"name": name, "id": user_id}
        )

# Тестирование с mock БД
class MockDatabase(Database):
    def query(self, sql: str) -> Any:
        return {"id": 1, "name": "Test User"}
    
    def insert(self, sql: str, params: Dict) -> int:
        return 999
    
    def update(self, sql: str, params: Dict) -> int:
        return 1

# Использование в тестах
def test_user_service():
    mock_db = MockDatabase()
    service = UserService(mock_db)
    
    user = service.get_user(1)
    assert user["name"] == "Test User"
    print("✅ Тест пройден")

# Использование в production
print("=== MySQL ===")
mysql_db = MySQLDatabase(host="localhost", user="root", password="pass")
user_service = UserService(mysql_db)
user = user_service.get_user(1)

print("\n=== PostgreSQL ===")
pg_db = PostgreSQLDatabase(host="localhost", user="postgres", password="pass")
user_service = UserService(pg_db)
user = user_service.get_user(1)

print("\n=== MongoDB ===")
mongo_db = MongoDBDatabase("mongodb://localhost:27017/mydb")
user_service = UserService(mongo_db)
user = user_service.get_user(1)

print("\n=== Тестирование ===")
test_user_service()

Пример 3: Логирование

from abc import ABC, abstractmethod
import logging

# Абстракция
class Logger(ABC):
    @abstractmethod
    def log(self, level: str, message: str) -> None:
        pass

# Конкретные реализации
class FileLogger(Logger):
    def __init__(self, filename: str):
        self.filename = filename
    
    def log(self, level: str, message: str) -> None:
        with open(self.filename, a) as f:
            f.write(f"[{level}] {message}\n")

class ConsoleLogger(Logger):
    def log(self, level: str, message: str) -> None:
        print(f"[{level}] {message}")

class SentryLogger(Logger):
    def log(self, level: str, message: str) -> None:
        # Отправить в Sentry
        if level == "ERROR":
            print(f"[SENTRY] Отправил ошибку: {message}")

# Бизнес-логика
class PaymentProcessor:
    def __init__(self, logger: Logger):
        self.logger = logger
    
    def process_payment(self, amount: float) -> bool:
        try:
            self.logger.log("INFO", f"Обработка платежа на сумму {amount}")
            # Обработка платежа
            self.logger.log("INFO", f"Платёж успешен")
            return True
        except Exception as e:
            self.logger.log("ERROR", f"Ошибка платежа: {str(e)}")
            return False

# Использование
print("=== Console Logging ===")
console_logger = ConsoleLogger()
processor = PaymentProcessor(console_logger)
processor.process_payment(100.0)

print("\n=== File Logging ===")
file_logger = FileLogger("/tmp/app.log")
processor = PaymentProcessor(file_logger)
processor.process_payment(50.0)

Пример 4: Dependency Injection Container

from abc import ABC, abstractmethod
from typing import Type, TypeVar, Dict, Any

T = TypeVar(T)

# Контейнер зависимостей
class DIContainer:
    def __init__(self):
        self._services: Dict[Type, Any] = {}
        self._factories: Dict[Type, callable] = {}
    
    def register(self, interface: Type[T], implementation: Type[T]) -> None:
        """Регистрирует реализацию для интерфейса"""
        self._services[interface] = implementation
    
    def register_factory(self, interface: Type[T], factory: callable) -> None:
        """Регистрирует фабрику для интерфейса"""
        self._factories[interface] = factory
    
    def resolve(self, interface: Type[T]) -> T:
        """Разрешает зависимость"""
        if interface in self._factories:
            return self._factories[interface]()
        
        if interface in self._services:
            implementation = self._services[interface]
            return implementation()
        
        raise ValueError(f"No registration for {interface}")

# Определяем интерфейсы
class NotificationService(ABC):
    @abstractmethod
    def send(self, recipient: str, message: str) -> bool:
        pass

class UserRepository(ABC):
    @abstractmethod
    def get_user(self, user_id: int):
        pass

# Реализации
class EmailNotificationService(NotificationService):
    def send(self, recipient: str, message: str) -> bool:
        print(f"📧 Email: {message}")
        return True

class MockUserRepository(UserRepository):
    def get_user(self, user_id: int):
        return {"id": user_id, "name": "Test User"}

# Бизнес-логика
class UserRegistrationService:
    def __init__(self, notification: NotificationService, repository: UserRepository):
        self.notification = notification
        self.repository = repository
    
    def register(self, user_id: int, email: str):
        user = self.repository.get_user(user_id)
        self.notification.send(email, f"Добро пожаловать, {user[name]}!")

# Конфигурация DI контейнера
container = DIContainer()
container.register(NotificationService, EmailNotificationService)
container.register(UserRepository, MockUserRepository)

# Использование
notification = container.resolve(NotificationService)
repository = container.resolve(UserRepository)
service = UserRegistrationService(notification, repository)
service.register(1, "user@example.com")

Преимущества DIP

# ✅ Тестирование
class TestUserService:
    def test_with_mock_db(self):
        mock_db = MockDatabase()
        service = UserService(mock_db)
        user = service.get_user(1)
        assert user is not None

# ✅ Расширяемость
# Можно добавить новую реализацию без изменения UserService
class RedisCache(Database):
    # Новая реализация
    pass

# ✅ Гибкость конфигурации
if config.get(USE_MYSQL):
    db = MySQLDatabase(...)
elif config.get(USE_POSTGRESQL):
    db = PostgreSQLDatabase(...)
else:
    db = MockDatabase()

service = UserService(db)

# ✅ Разделение ответственности
# UserService отвечает за бизнес-логику
# Database - за работу с хранилищем
# Они не знают о деталях реализации друг друга

Антипаттерны

# ❌ НЕПРАВИЛЬНО: new внутри класса
class BadService:
    def __init__(self):
        self.db = MySQLDatabase()  # Жёсткая зависимость!
        self.logger = FileLogger()  # Жёсткая зависимость!
        self.cache = RedisCache()   # Жёсткая зависимость!

# ✅ ПРАВИЛЬНО: инъекция зависимостей
class GoodService:
    def __init__(
        self,
        db: Database,
        logger: Logger,
        cache: Cache
    ):
        self.db = db
        self.logger = logger
        self.cache = cache

Заключение

Dependency Inversion Principle помогает создавать:

• Тестируемый код — легко подменять реальные сервисы на mock-и
• Гибкий код — просто менять реализации
• Слабосвязанный код — модули не знают друг о друге
• Масштабируемый код — легко добавлять новые реализации

Ключевая идея: зависимости должны указывать вверх (от конкретного к абстрактному), а не вниз.