Singleton, Factory, Strategy, Observer, Decorator, Adapter, Facade, Builder в контексте Python
Python — динамический язык: многие классические паттерны ГОФ реализуются через функции, декораторы, duck typing. Понимайте проблему, а не механическое применение паттерна.
| Категория | Назначение |
|---|---|
| Порождающие | Создание объектов (Singleton, Factory, Builder, Prototype) |
| Структурные | Компоновка объектов (Adapter, Decorator, Facade, Proxy) |
| Поведенческие | Взаимодействие объектов (Strategy, Observer, Command, Template Method) |
import threading
class SingletonMeta(type):
_instances = {}
_lock = threading.Lock()
def __call__(cls, *args, **kwargs):
with cls._lock: # thread-safe
if cls not in cls._instances:
cls._instances[cls] = super().__call__(*args, **kwargs)
return cls._instances[cls]
class Config(metaclass=SingletonMeta):
def __init__(self):
self.settings = {}
def load(self, path):
import json
with open(path) as f:
self.settings = json.load(f)
c1 = Config()
c2 = Config()
assert c1 is c2 # True__new__class Singleton:
_instance = None
def __new__(cls, *args, **kwargs):
if cls._instance is None:
cls._instance = super().__new__(cls)
return cls._instance# config.py
_settings = {}
def get_settings():
return _settings
def load(path):
import json
with open(path) as f:
_settings.update(json.load(f))
# Модуль загружается один раз — это и есть singleton
# import config; config.load("settings.json"); config.get_settings()from abc import ABC, abstractmethod
class Notification(ABC):
@abstractmethod
def send(self, message: str) -> None: ...
class EmailNotification(Notification):
def __init__(self, email: str):
self.email = email
def send(self, message: str) -> None:
print(f"Email to {self.email}: {message}")
class SMSNotification(Notification):
def __init__(self, phone: str):
self.phone = phone
def send(self, message: str) -> None:
print(f"SMS to {self.phone}: {message}")
class PushNotification(Notification):
def __init__(self, device_id: str):
self.device_id = device_id
def send(self, message: str) -> None:
print(f"Push to {self.device_id}: {message}")
# Фабричная функция — проще чем Factory Method класс
def create_notification(kind: str, **kwargs) -> Notification:
factories = {
"email": EmailNotification,
"sms": SMSNotification,
"push": PushNotification,
}
if kind not in factories:
raise ValueError(f"Unknown notification type: {kind!r}")
return factories[kind](**kwargs)
# Использование
notif = create_notification("email", email="alice@example.com")
notif.send("Hello!")__init_subclass__class PluginBase:
_registry: dict[str, type] = {}
def __init_subclass__(cls, plugin_name: str = "", **kwargs):
super().__init_subclass__(**kwargs)
if plugin_name:
PluginBase._registry[plugin_name] = cls
@classmethod
def create(cls, name: str, **kwargs) -> "PluginBase":
if name not in cls._registry:
raise KeyError(f"Unknown plugin: {name!r}")
return cls._registry[name](**kwargs)
class CSVExporter(PluginBase, plugin_name="csv"):
def export(self, data): return ",".join(str(x) for x in data)
class JSONExporter(PluginBase, plugin_name="json"):
def export(self, data):
import json
return json.dumps(data)
# Новые плагины регистрируются автоматически при определении класса
exporter = PluginBase.create("csv")
print(exporter.export([1, 2, 3])) # "1,2,3"from dataclasses import dataclass, field
from typing import Optional
@dataclass
class Query:
table: str
conditions: list = field(default_factory=list)
columns: list = field(default_factory=list)
_limit: Optional[int] = None
_offset: int = 0
_order_by: Optional[str] = None
def to_sql(self) -> str:
cols = ", ".join(self.columns) if self.columns else "*"
sql = f"SELECT {cols} FROM {self.table}"
if self.conditions:
sql += " WHERE " + " AND ".join(self.conditions)
if self._order_by:
sql += f" ORDER BY {self._order_by}"
if self._limit is not None:
sql += f" LIMIT {self._limit} OFFSET {self._offset}"
return sql
class QueryBuilder:
def __init__(self, table: str):
self._query = Query(table)
def select(self, *columns: str) -> "QueryBuilder":
self._query.columns = list(columns)
return self # возвращаем self для chaining
def where(self, condition: str) -> "QueryBuilder":
self._query.conditions.append(condition)
return self
def limit(self, n: int, offset: int = 0) -> "QueryBuilder":
self._query._limit = n
self._query._offset = offset
return self
def order_by(self, column: str) -> "QueryBuilder":
self._query._order_by = column
return self
def build(self) -> Query:
return self._query
# Fluent interface (method chaining)
query = (
QueryBuilder("users")
.select("id", "name", "email")
.where("age > 18")
.where("active = true")
.order_by("name")
.limit(10)
.build()
)
print(query.to_sql())
# SELECT id, name, email FROM users WHERE age > 18 AND active = true ORDER BY name LIMIT 10 OFFSET 0import copy
from typing import TypeVar
T = TypeVar("T", bound="Prototype")
class Prototype:
def clone(self: T) -> T:
"""Поверхностная копия."""
return copy.copy(self)
def deep_clone(self: T) -> T:
"""Глубокая копия."""
return copy.deepcopy(self)
class DocumentTemplate(Prototype):
def __init__(self, title: str, content: str, metadata: dict):
self.title = title
self.content = content
self.metadata = metadata # изменяемый объект!
def __repr__(self):
return f"Document({self.title!r})"
template = DocumentTemplate(
"Report Template",
"## {title}\n\n{body}",
{"author": "default", "version": 1}
)
# Быстрое создание похожих объектов
report1 = template.deep_clone()
report1.title = "Q1 Report"
report1.metadata["author"] = "Alice"
report2 = template.deep_clone()
report2.title = "Q2 Report"
report2.metadata["author"] = "Bob"
# template.metadata["author"] остался "default"В Python функции — first-class objects, поэтому Strategy часто реализуется без классов.
from typing import Callable, TypeVar
T = TypeVar("T")
# Стратегия как функция — самый Pythonic способ
def sort_by_name(items: list) -> list:
return sorted(items, key=lambda x: x["name"])
def sort_by_age(items: list) -> list:
return sorted(items, key=lambda x: x["age"])
def sort_by_score_desc(items: list) -> list:
return sorted(items, key=lambda x: x["score"], reverse=True)
class ReportGenerator:
def __init__(self, sort_strategy: Callable[[list], list]):
self._sort = sort_strategy
def generate(self, data: list) -> str:
sorted_data = self._sort(data)
return "\n".join(f"{d['name']}: {d['score']}" for d in sorted_data)
def set_strategy(self, strategy: Callable[[list], list]) -> None:
self._sort = strategy
data = [
{"name": "Alice", "age": 30, "score": 85},
{"name": "Charlie", "age": 25, "score": 92},
{"name": "Bob", "age": 35, "score": 78},
]
gen = ReportGenerator(sort_by_name)
print(gen.generate(data)) # Alice, Bob, Charlie
gen.set_strategy(sort_by_score_desc)
print(gen.generate(data)) # Charlie, Alice, Bobfrom typing import Callable, Any
from collections import defaultdict
class EventBus:
"""Простая шина событий."""
def __init__(self):
self._handlers: dict[str, list[Callable]] = defaultdict(list)
def subscribe(self, event: str, handler: Callable) -> None:
self._handlers[event].append(handler)
def unsubscribe(self, event: str, handler: Callable) -> None:
self._handlers[event].remove(handler)
def publish(self, event: str, **data: Any) -> None:
for handler in self._handlers[event]:
handler(**data)
bus = EventBus()
# Подписчики
def send_welcome_email(user_id, email, **_):
print(f"Sending welcome to {email}")
def update_analytics(user_id, **_):
print(f"Tracking registration for user {user_id}")
def create_default_settings(user_id, **_):
print(f"Creating settings for user {user_id}")
bus.subscribe("user.registered", send_welcome_email)
bus.subscribe("user.registered", update_analytics)
bus.subscribe("user.registered", create_default_settings)
# Публикация события
bus.publish("user.registered", user_id=42, email="alice@example.com")
# Все три хендлера вызовутсяclass Observable:
"""Базовый класс с поддержкой наблюдаемых свойств."""
def __init__(self):
self._observers: dict[str, list] = defaultdict(list)
def on(self, event: str):
"""Декоратор для подписки на событие."""
def decorator(func):
self._observers[event].append(func)
return func
return decorator
def emit(self, event: str, *args, **kwargs):
for handler in self._observers[event]:
handler(*args, **kwargs)
class UserService(Observable):
def __init__(self):
super().__init__()
self._users = {}
def create_user(self, name: str, email: str) -> dict:
user = {"id": len(self._users) + 1, "name": name, "email": email}
self._users[user["id"]] = user
self.emit("created", user=user)
return user
service = UserService()
@service.on("created")
def on_user_created(user):
print(f"New user: {user['name']}")
service.create_user("Alice", "alice@example.com")from abc import ABC, abstractmethod
from typing import Optional
class Command(ABC):
@abstractmethod
def execute(self) -> None: ...
def undo(self) -> None:
raise NotImplementedError("This command is not undoable")
class TextEditor:
def __init__(self):
self.text = ""
def insert(self, pos: int, text: str) -> None:
self.text = self.text[:pos] + text + self.text[pos:]
def delete(self, start: int, end: int) -> None:
self.text = self.text[:start] + self.text[end:]
class InsertCommand(Command):
def __init__(self, editor: TextEditor, pos: int, text: str):
self.editor = editor
self.pos = pos
self.text = text
def execute(self) -> None:
self.editor.insert(self.pos, self.text)
def undo(self) -> None:
self.editor.delete(self.pos, self.pos + len(self.text))
class CommandHistory:
def __init__(self):
self._history: list[Command] = []
self._redo_stack: list[Command] = []
def execute(self, command: Command) -> None:
command.execute()
self._history.append(command)
self._redo_stack.clear() # redo стак очищается при новой команде
def undo(self) -> Optional[Command]:
if not self._history:
return None
command = self._history.pop()
command.undo()
self._redo_stack.append(command)
return command
def redo(self) -> Optional[Command]:
if not self._redo_stack:
return None
command = self._redo_stack.pop()
command.execute()
self._history.append(command)
return command
editor = TextEditor()
history = CommandHistory()
history.execute(InsertCommand(editor, 0, "Hello"))
history.execute(InsertCommand(editor, 5, " World"))
print(editor.text) # "Hello World"
history.undo()
print(editor.text) # "Hello"
history.redo()
print(editor.text) # "Hello World"from abc import ABC, abstractmethod
class DataProcessor(ABC):
"""Шаблонный метод задаёт порядок шагов."""
def process(self, source: str) -> list:
"""Алгоритм обработки данных."""
raw_data = self._read(source)
parsed = self._parse(raw_data)
validated = [item for item in parsed if self._validate(item)]
return [self._transform(item) for item in validated]
@abstractmethod
def _read(self, source: str) -> str:
"""Чтение данных (переопределяется в подклассах)."""
...
@abstractmethod
def _parse(self, data: str) -> list:
...
def _validate(self, item) -> bool:
return True # по умолчанию всё валидно
def _transform(self, item):
return item # по умолчанию без изменений
class CSVProcessor(DataProcessor):
def _read(self, source: str) -> str:
with open(source) as f:
return f.read()
def _parse(self, data: str) -> list:
import csv, io
reader = csv.DictReader(io.StringIO(data))
return list(reader)
def _validate(self, item) -> bool:
return bool(item.get("name"))
class JSONAPIProcessor(DataProcessor):
def _read(self, source: str) -> str:
import requests
return requests.get(source).text
def _parse(self, data: str) -> list:
import json
return json.loads(data)["items"]
def _transform(self, item) -> dict:
return {"id": item["id"], "title": item["title"].strip()}# Старый API (изменить нельзя — legacy)
class LegacyPaymentGateway:
def charge(self, card_number: str, amount_cents: int, currency: str) -> bool:
print(f"Charging {amount_cents} {currency} to {card_number[-4:]}")
return True
# Новый интерфейс (нужен в нашем коде)
class ModernPaymentGateway(ABC):
@abstractmethod
def process_payment(self, payment_data: dict) -> dict:
"""payment_data: {"card": "...", "amount": 9.99, "currency": "USD"}"""
...
# Адаптер: реализует новый интерфейс, использует старый внутри
class LegacyPaymentAdapter(ModernPaymentGateway):
def __init__(self, legacy: LegacyPaymentGateway):
self._legacy = legacy
def process_payment(self, payment_data: dict) -> dict:
amount_cents = int(payment_data["amount"] * 100)
success = self._legacy.charge(
payment_data["card"],
amount_cents,
payment_data.get("currency", "USD")
)
return {"success": success, "amount": payment_data["amount"]}
# Адаптер функции — ещё проще
def adapt_legacy_payment(legacy_gateway):
def process_payment(payment_data: dict) -> dict:
amount_cents = int(payment_data["amount"] * 100)
success = legacy_gateway.charge(
payment_data["card"], amount_cents, payment_data.get("currency", "USD")
)
return {"success": success}
return process_paymentОтличается от Python-декораторов. Обёртывает объект, не функцию.
from abc import ABC, abstractmethod
class Logger(ABC):
@abstractmethod
def log(self, message: str) -> None: ...
class ConsoleLogger(Logger):
def log(self, message: str) -> None:
print(message)
# Базовый декоратор
class LoggerDecorator(Logger):
def __init__(self, logger: Logger):
self._logger = logger
def log(self, message: str) -> None:
self._logger.log(message)
# Конкретные декораторы
class TimestampDecorator(LoggerDecorator):
def log(self, message: str) -> None:
from datetime import datetime
self._logger.log(f"[{datetime.now().isoformat()}] {message}")
class LevelDecorator(LoggerDecorator):
def __init__(self, logger: Logger, level: str = "INFO"):
super().__init__(logger)
self.level = level
def log(self, message: str) -> None:
self._logger.log(f"[{self.level}] {message}")
class FilterDecorator(LoggerDecorator):
def __init__(self, logger: Logger, min_length: int = 0):
super().__init__(logger)
self.min_length = min_length
def log(self, message: str) -> None:
if len(message) >= self.min_length:
self._logger.log(message)
# Составной логгер через декорирование
logger = (
FilterDecorator(
LevelDecorator(
TimestampDecorator(ConsoleLogger()),
level="ERROR"
),
min_length=5
)
)
logger.log("Hi") # отфильтровано (< 5 символов)
logger.log("Something bad happened") # [ERROR] [2024-01-01T...] Something bad happenedclass EmailSender:
def connect(self, host: str, port: int): ...
def authenticate(self, user: str, password: str): ...
def send(self, to: str, subject: str, body: str): ...
def disconnect(self): ...
class SMSProvider:
def initialize(self, api_key: str): ...
def send_sms(self, phone: str, text: str): ...
class PushService:
def register_device(self, token: str): ...
def send_push(self, token: str, title: str, body: str): ...
# Фасад скрывает сложность подсистемы
class NotificationFacade:
def __init__(self, config: dict):
self._email = EmailSender()
self._sms = SMSProvider()
self._push = PushService()
self._config = config
def _setup_email(self):
self._email.connect(self._config["smtp_host"], self._config["smtp_port"])
self._email.authenticate(self._config["smtp_user"], self._config["smtp_pass"])
def notify_user(
self,
user: dict,
message: str,
channels: list[str] = ("email",)
) -> None:
if "email" in channels and user.get("email"):
self._setup_email()
self._email.send(user["email"], "Notification", message)
self._email.disconnect()
if "sms" in channels and user.get("phone"):
self._sms.initialize(self._config["sms_key"])
self._sms.send_sms(user["phone"], message)
if "push" in channels and user.get("device_token"):
self._push.register_device(user["device_token"])
self._push.send_push(user["device_token"], "Update", message)
# Клиентский код прост:
notifier = NotificationFacade(config)
notifier.notify_user(user, "Your order shipped!", channels=["email", "push"])| Антипаттерн | Проблема |
|---|---|
| Singleton везде | Глобальное состояние → трудно тестировать |
| Factory для всего | Усложняет без пользы если классов ≤ 2-3 |
Decorator вместо @functools.wraps | Python-декораторы проще |
| Observer без ограничений | Утечки памяти если подписчики не отписались |
| Command без undo | Overhead без пользы |
| Template Method вместо Protocol | Protocol + функции часто гибче |
IDLE, HAS_COIN, DISPENSING. Переходы: insert_coin(), turn_knob(), dispense()..build() возвращает requests.Request.DEBUG → INFO → WARNING → ERROR — каждый обрабатывает свой уровень и передаёт дальше.__init_subclass__ для системы валидаторов: @validator("email"), @validator("phone") автоматически регистрируют функции.Вопросы ещё не добавлены
Вопросы для этой подтемы ещё не добавлены.