Паттерны проектирования

Паттерны проектирования в Python

Маршрут: Инженерная основа · тема 8 из 8

Навыки: M1, M2, S6

До урока: functions, oop, dataclasses, closures_decorators, type_system.

Результат: применение паттерна только там, где он проще наблюдаемой альтернативы.

Основа: «Стратегия», «Фабрика», «Адаптер» и «Наблюдатель». Работа в эксплуатации: компромиссы, расширяемость и отказ от лишней церемонии. Углубление: варианты на метаклассах; вернитесь после metaclasses.

Подтверждение: реализация GoF и второй, более простой рефакторинг в стиле Python.

В Python часть классических GoF-паттернов выражается обычной функцией, декоратором или структурной типизацией. Выбирайте паттерн по изменчивости и контракту задачи, а не по названию.

#1. Singleton — единственный экземпляр

#Через метакласс (thread-safe)

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

#Через модуль (самый Pythonic)

# 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()

#2. Factory — создание объектов без привязки к классу

#Простая фабрика (функция)

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!")

#Registry Pattern — саморегистрация через `__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"

#3. Builder — пошаговое создание сложных объектов

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 0

#4. Prototype — клонирование объектов

import 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"

#5. Strategy — взаимозаменяемые алгоритмы

В 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, Bob

#6. Observer — события и подписчики

from 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")
# Все три хендлера вызовутся

#Observable property

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")

#7. Command — инкапсуляция операций

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"

#8. Template Method — алгоритм с переопределяемыми шагами

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()}

#9. Adapter — адаптация интерфейса

# Старый 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

#10. Decorator (паттерн) — расширение поведения

Отличается от 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 happened

#11. Facade — упрощённый интерфейс

class 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"])

#12. Chain of Responsibility — цепочка обработчиков

Цепочка передаёт запрос обработчикам по порядку. Каждый обработчик либо завершает обработку, либо явно передаёт запрос дальше. Паттерн подходит, когда состав и порядок шагов настраиваются, но обычного цикла по функциям бывает достаточно.

from collections.abc import Callable

Handler = Callable[[dict], bool]


def run_chain(request: dict, handlers: list[Handler]) -> bool:
    for handler in handlers:
        if handler(request):
            return True
    return False

Здесь True означает «запрос обработан». Контракт остановки нужно назвать явно, иначе часть цепочки легко пропустить случайно.

#13. State — поведение зависит от состояния

State выносит поведение каждого состояния в отдельный объект. Контекст делегирует ему операцию и хранит текущее состояние. Это убирает большой if/elif, когда переходов много и у каждого состояния разные допустимые действия.

from typing import Protocol


class State(Protocol):
    def handle(self, context: "Order") -> None: ...


class Draft:
    def handle(self, context: "Order") -> None:
        context.state = Paid()


class Paid:
    def handle(self, context: "Order") -> None:
        context.state = Shipped()


class Shipped:
    def handle(self, context: "Order") -> None:
        raise RuntimeError("Заказ уже отправлен")


class Order:
    def __init__(self) -> None:
        self.state: State = Draft()

    def advance(self) -> None:
        self.state.handle(self)

Если состояний два и переходы просты, перечисление и обычный match обычно понятнее отдельной иерархии классов.

#14. Когда не использовать паттерны

Антипаттерн	Проблема
Singleton везде	Глобальное состояние → трудно тестировать
Factory для всего	Усложняет без пользы если классов ≤ 2-3
Decorator вместо `@functools.wraps`	Python-декораторы проще
Observer без ограничений	Утечки памяти если подписчики не отписались
Command без undo	Overhead без пользы
Template Method вместо Protocol	Protocol + функции часто гибче

#Категории паттернов

Категория	Назначение
Порождающие	Создание объектов (Singleton, Factory, Builder, Prototype)
Структурные	Компоновка объектов (Adapter, Decorator, Facade, Proxy)
Поведенческие	Взаимодействие объектов (Strategy, Observer, Command, Template Method)

#Практика: от пробы к рабочему решению

#Шаг 1. Соберите минимальный пример

Начинайте с наблюдаемой изменчивости задачи, а не с названия паттерна. Результат основы — две реализации одного контракта и критерий, когда более сложная из них окупается.

#Шаг 2. Проверьте поведение

from collections.abc import Callable

Discount = Callable[[int], int]

def regular(total: int) -> int:
    return total

def vip(total: int) -> int:
    return round(total * 0.9)

strategies: dict[str, Discount] = {"regular": regular, "vip": vip}
assert strategies["regular"](1000) == 1000
assert strategies["vip"](1000) == 900

#Шаг 3. Доведите решение до рабочего сценария

Практика с подсказками: добавьте стратегию из конфигурации, неизвестное имя и наблюдаемую доменную ошибку. Затем реализуйте объектный вариант и сравните расширяемость, тестируемость и стоимость регистрации.

#Шаг 4. Объясните внутренний механизм

После урока о метаклассах соберите автоматический реестр классов, а затем замените его явным словарём, декоратором или __init_subclass__. Защитите метакласс только если более простые варианты не выражают нужный контракт создания класса.

#Упражнения

Реализуйте паттерн State для торгового автомата: состояния IDLE, HAS_COIN, DISPENSING. Переходы: insert_coin(), turn_knob(), dispense().
Напишите Builder для HTTP-запроса: метод (GET/POST), URL, заголовки, параметры, тело. .build() возвращает requests.Request.
Создайте Event System с приоритетами подписчиков — обработчики с высоким приоритетом вызываются первыми.
Реализуйте Chain of Responsibility: обработчики логов DEBUG → INFO → WARNING → ERROR — каждый обрабатывает свой уровень и передаёт дальше.
Реализуйте Registry через __init_subclass__ для системы валидаторов: @validator("email"), @validator("phone") автоматически регистрируют функции.

#Самопроверка

Для State составьте таблицу всех допустимых и запрещённых переходов до кода; каждый запрещённый переход даёт определённую ошибку и не меняет состояние.
HTTP Builder валидирует схему URL, не разделяет изменяемые headers между сборками и возвращает новый Request.
Event System определяет порядок для одинакового приоритета и политику после ошибки обработчика.
Chain не должен обрабатывать одну запись несколько раз без явного решения; проверьте неизвестный уровень.
Registry отклоняет дубликаты и позволяет заменить декораторы обычным словарём; объясните, зачем выбран более сложный механизм.

#Практическая лаборатория

Закрепите тему в лаборатории «Маршрутизатор стратегий без лишней церемонии». Вы реализуете объектную и функциональную стратегию с одним контрактом и защитите более простой вариант там, где отдельный класс не даёт пользы.

git clone --branch v0.6.2 --depth 1 https://gitlab.potapov.me/courses/python-labs.git
cd python-labs
uv sync --group test
uv run --group test pytest patterns/tests

Антипаттерн

Проблема

Singleton везде

Глобальное состояние → трудно тестировать

Factory для всего

Усложняет без пользы если классов ≤ 2-3

Decorator вместо @functools.wraps

Python-декораторы проще

Observer без ограничений

Утечки памяти если подписчики не отписались

Command без undo

Overhead без пользы

Template Method вместо Protocol

Protocol + функции часто гибче

Категория

Назначение

Порождающие

Создание объектов (Singleton, Factory, Builder, Prototype)

Структурные

Компоновка объектов (Adapter, Decorator, Facade, Proxy)

Поведенческие

Взаимодействие объектов (Strategy, Observer, Command, Template Method)