Семейство алгоритмов через стратегию. Dependency Injection через конструктор и инжекторы. dependency_injector.
Strategy выбирает алгоритм. DI предоставляет зависимости.
Strategy — поведенческий паттерн, определяющий семейство алгоритмов, инкапсулирующий каждый из них и делающий их взаимозаменяемыми.
# ❌ ПЛОХО: Много if/elif для алгоритмов
class Order:
def __init__(self, items: list, shipping_method: str):
self.items = items
self.shipping_method = shipping_method
def calculate_shipping(self) -> float:
if self.shipping_method == "ground":
return sum(item.weight for item in self.items) * 0.5
elif self.shipping_method == "air":
return sum(item.weight for item in self.items) * 1.5
elif self.shipping_method == "express":
return sum(item.weight for item in self.items) * 2.5 + 10
else:
raise ValueError(f"Unknown method: {self.shipping_method}")
# Проблема:
# 1. Нарушен Open-Closed — добавление метода требует изменения кода
# 2. Логика размазана по условиям
# 3. Трудно тестировать отдельные алгоритмы# ✅ ХОРОШО: Каждый алгоритм — отдельный класс
from abc import ABC, abstractmethod
class ShippingStrategy(ABC):
@abstractmethod
def calculate(self, items: list) -> float:
pass
class GroundShipping(ShippingStrategy):
def calculate(self, items: list) -> float:
return sum(item.weight for item in items) * 0.5
class AirShipping(ShippingStrategy):
def calculate(self, items: list) -> float:
return sum(item.weight for item in items) * 1.5
class ExpressShipping(ShippingStrategy):
def calculate(self, items: list) -> float:
return sum(item.weight for item in items) * 2.5 + 10
class Order:
def __init__(self, items: list, strategy: ShippingStrategy):
self.items = items
self._strategy = strategy
@property
def strategy(self) -> ShippingStrategy:
return self._strategy
@strategy.setter
def strategy(self, strategy: ShippingStrategy):
self._strategy = strategy
def calculate_shipping(self) -> float:
return self._strategy.calculate(self.items)
# Использование
items = [Item(1), Item(2), Item(3)] # weight: 1, 2, 3
order = Order(items, GroundShipping())
print(order.calculate_shipping()) # 3.0
# Можно изменить стратегию в runtime
order.strategy = AirShipping()
print(order.calculate_shipping()) # 9.0
order.strategy = ExpressShipping()
print(order.calculate_shipping()) # 25.0# ✅ Проще: функции как стратегии
from typing import Callable, List
class Item:
def __init__(self, weight: float):
self.weight = weight
class Order:
def __init__(self, items: List[Item], strategy: Callable[[List[Item]], float]):
self.items = items
self._strategy = strategy
@property
def strategy(self) -> Callable[[List[Item]], float]:
return self._strategy
@strategy.setter
def strategy(self, strategy: Callable[[List[Item]], float]):
self._strategy = strategy
def calculate_shipping(self) -> float:
return self._strategy(self.items)
# Стратегии как функции
def ground_shipping(items: List[Item]) -> float:
return sum(item.weight for item in items) * 0.5
def air_shipping(items: List[Item]) -> float:
return sum(item.weight for item in items) * 1.5
def express_shipping(items: List[Item]) -> float:
return sum(item.weight for item in items) * 2.5 + 10
# Использование
items = [Item(1), Item(2), Item(3)]
order = Order(items, ground_shipping)
print(order.calculate_shipping()) # 3.0
order.strategy = air_shipping
print(order.calculate_shipping()) # 9.0# ✅ DI через конструктор
class Logger:
def log(self, message: str):
print(f"[LOG] {message}")
class Database:
def __init__(self, logger: Logger):
self._logger = logger
def query(self, sql: str):
self._logger.log(f"Executing: {sql}")
return []
# Использование
logger = Logger()
db = Database(logger) # Зависимость внедряется извне
db.query("SELECT * FROM users")# ✅ DI через параметры метода
class PaymentService:
def process(self, amount: float, gateway):
"""gateway внедряется при вызове"""
self._logger.log(f"Processing ${amount}")
return gateway.charge(amount)
# Использование
payment = PaymentService()
# В production
payment.process(100, StripeGateway())
# В тестах
payment.process(100, MockGateway())# ✅ Простой DI контейнер
from typing import Type, Dict, Callable, Any
class Container:
def __init__(self):
self._services: Dict[str, Callable] = {}
self._instances: Dict[str, Any] = {}
def register(self, name: str, factory: Callable, singleton: bool = True):
self._services[name] = (factory, singleton)
def get(self, name: str) -> Any:
if name not in self._services:
raise KeyError(f"Service not found: {name}")
factory, singleton = self._services[name]
if singleton:
if name not in self._instances:
self._instances[name] = factory()
return self._instances[name]
return factory()
def __getitem__(self, name: str) -> Any:
return self.get(name)
# Использование
container = Container()
container.register("logger", lambda: Logger())
container.register("database", lambda: Database(container["logger"]))
container.register("payment", lambda: PaymentService(container["logger"]))
# Получение сервисов
db = container["database"]
payment = container["payment"]# ✅ Стратегии для сортировки
from typing import List, Callable, TypeVar
T = TypeVar('T')
class Sorter:
def __init__(self, strategy: Callable[[List[T]], List[T]]):
self._strategy = strategy
@property
def strategy(self) -> Callable[[List[T]], List[T]]:
return self._strategy
@strategy.setter
def strategy(self, strategy: Callable[[List[T]], List[T]]):
self._strategy = strategy
def sort(self, data: List[T]) -> List[T]:
return self._strategy(data)
# Стратегии
def bubble_sort(data: List[T]) -> List[T]:
print("Bubble sort")
arr = data.copy()
n = len(arr)
for i in range(n):
for j in range(n - i - 1):
if arr[j] > arr[j + 1]:
arr[j], arr[j + 1] = arr[j + 1], arr[j]
return arr
def quick_sort(data: List[T]) -> List[T]:
print("Quick sort")
if len(data) <= 1:
return data
pivot = data[len(data) // 2]
left = [x for x in data if x < pivot]
middle = [x for x in data if x == pivot]
right = [x for x in data if x > pivot]
return quick_sort(left) + middle + quick_sort(right)
# Использование
sorter = Sorter(bubble_sort)
print(sorter.sort([3, 1, 4, 1, 5]))
sorter.strategy = quick_sort
print(sorter.sort([3, 1, 4, 1, 5]))
# Или встроенная стратегия
sorter.strategy = sorted
print(sorter.sort([3, 1, 4, 1, 5]))# ✅ Стратегии аутентификации
from abc import ABC, abstractmethod
from typing import Optional
class AuthStrategy(ABC):
@abstractmethod
def authenticate(self, credentials: dict) -> Optional[dict]:
pass
class PasswordAuth(AuthStrategy):
def authenticate(self, credentials: dict) -> Optional[dict]:
username = credentials.get("username")
password = credentials.get("password")
# Проверка пароля в БД
if username and password:
return {"user": username, "method": "password"}
return None
class TokenAuth(AuthStrategy):
def __init__(self, secret: str):
self._secret = secret
def authenticate(self, credentials: dict) -> Optional[dict]:
token = credentials.get("token")
# Проверка JWT токена
if token and token.startswith("valid_"):
return {"user": "token_user", "method": "token"}
return None
class OAuthAuth(AuthStrategy):
def __init__(self, provider: str):
self._provider = provider
def authenticate(self, credentials: dict) -> Optional[dict]:
code = credentials.get("code")
# OAuth flow
if code:
return {"user": "oauth_user", "provider": self._provider}
return None
class AuthService:
def __init__(self, strategy: AuthStrategy):
self._strategy = strategy
@property
def strategy(self) -> AuthStrategy:
return self._strategy
@strategy.setter
def strategy(self, strategy: AuthStrategy):
self._strategy = strategy
def login(self, credentials: dict) -> Optional[dict]:
return self._strategy.authenticate(credentials)
# Использование
auth = AuthService(PasswordAuth())
user = auth.login({"username": "alice", "password": "secret"})
auth.strategy = TokenAuth("secret_key")
user = auth.login({"token": "valid_abc123"})# ✅ DI для легкого тестирования
class EmailSender:
def send(self, to: str, subject: str, body: str):
# Реальная отправка email
print(f"Sending email to {to}")
class MockEmailSender:
def __init__(self):
self.sent_emails = []
def send(self, to: str, subject: str, body: str):
self.sent_emails.append({
"to": to,
"subject": subject,
"body": body,
})
class UserService:
def __init__(self, db, email_sender: EmailSender):
self._db = db
self._email_sender = email_sender
def register(self, username: str, email: str):
# Сохранение в БД
self._email_sender.send(
email,
"Welcome!",
f"Hello {username}!"
)
# В production
user_service = UserService(Database(), EmailSender())
# В тестах
mock_email = MockEmailSender()
user_service = UserService(MockDatabase(), mock_email)
user_service.register("alice", "alice@example.com")
assert len(mock_email.sent_emails) == 1
assert mock_email.sent_emails[0]["to"] == "alice@example.com"# ✅ dependency_injector — готовый DI фреймворк
# pip install dependency_injector
from dependency_injector import containers, providers
from dependency_injector.wiring import Provide, inject
class Container(containers.DeclarativeContainer):
config = providers.Configuration()
logger = providers.Singleton(Logger)
database = providers.Singleton(Database, logger=logger)
payment = providers.Singleton(PaymentService, logger=logger)
# Использование
container = Container()
container.config.from_dict({"debug": True})
# Получение сервисов
db = container.database()
payment = container.payment()
# Или с wiring
class Application:
@inject
def __init__(self, db: Database = Provide[Container.database]):
self._db = dbИзучите тему Template Method и Hooks для скелета алгоритма.
Далее: Template Method и Hooks