Паттерны проектирования в Python: боевые примеры для надежного кода

# Реализация через метакласс
class SingletonMeta(type):
_instances = {}

def __call__(cls, *args, **kwargs):
if cls not in cls._instances:
cls._instances[cls] = super().__call__(*args, **kwargs)
return cls._instances[cls]

class DatabaseConnection(metaclass=SingletonMeta):
def __init__(self):
self.connection = "Established"
print("Connection created")

# При многократном вызове будет возвращаться один и тот же экземпляр
db1 = DatabaseConnection() # Вывод: "Connection created"
db2 = DatabaseConnection() # Без вывода, т.к. используется существующий экземпляр
print(db1 is db2) # Вывод: True

# Пример фабричного метода
from abc import ABC, abstractmethod

class Animal(ABC):
@abstractmethod
def speak(self):
pass

class Dog(Animal):
def speak(self):
return "Woof!"

class Cat(Animal):
def speak(self):
return "Meow!"

class AnimalFactory:
def create_animal(self, animal_type):
if animal_type == "dog":
return Dog()
elif animal_type == "cat":
return Cat()
else:
raise ValueError(f"Unknown animal type: {animal_type}")

factory = AnimalFactory()
dog = factory.create_animal("dog")
print(dog.speak()) # Вывод: "Woof!"

# Пример абстрактной фабрики
from abc import ABC, abstractmethod

# Абстрактные продукты
class Button(ABC):
@abstractmethod
def render(self):
pass

class Checkbox(ABC):
@abstractmethod
def render(self):
pass

# Конкретные продукты для Windows
class WindowsButton(Button):
def render(self):
return "Windows Button"

class WindowsCheckbox(Checkbox):
def render(self):
return "Windows Checkbox"

# Конкретные продукты для MacOS
class MacOSButton(Button):
def render(self):
return "MacOS Button"

class MacOSCheckbox(Checkbox):
def render(self):
return "MacOS Checkbox"

# Абстрактная фабрика
class GUIFactory(ABC):
@abstractmethod
def create_button(self):
pass

@abstractmethod
def create_checkbox(self):
pass

# Конкретные фабрики
class WindowsFactory(GUIFactory):
def create_button(self):
return WindowsButton()

def create_checkbox(self):
return WindowsCheckbox()

class MacOSFactory(GUIFactory):
def create_button(self):
return MacOSButton()

def create_checkbox(self):
return MacOSCheckbox()

# Клиентский код
def create_ui(factory):
button = factory.create_button()
checkbox = factory.create_checkbox()
return f"{button.render()} and {checkbox.render()}"

windows_ui = create_ui(WindowsFactory())
print(windows_ui) # Вывод: "Windows Button and Windows Checkbox"

# Существующий класс с несовместимым интерфейсом
class LegacyRectangle:
def __init__(self, x1, y1, x2, y2):
self.x1 = x1
self.y1 = y1
self.x2 = x2
self.y2 = y2

# Целевой интерфейс
class Rectangle:
def __init__(self, x, y, width, height):
self.x = x
self.y = y
self.width = width
self.height = height

# Адаптер
class RectangleAdapter(Rectangle):
def __init__(self, legacy_rectangle):
width = legacy_rectangle.x2 – legacy_rectangle.x1
height = legacy_rectangle.y2 – legacy_rectangle.y1
super().__init__(legacy_rectangle.x1, legacy_rectangle.y1, width, height)

# Клиентский код
legacy_rect = LegacyRectangle(1, 1, 5, 5)
adapted_rect = RectangleAdapter(legacy_rect)
print(f"Position: ({adapted_rect.x}, {adapted_rect.y}), Size: {adapted_rect.width}x{adapted_rect.height}")

# Пример декоратора функции
import functools
import time

def timing_decorator(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
start_time = time.time()
result = func(*args, **kwargs)
end_time = time.time()
print(f"Функция {func.__name__} выполнилась за {end_time – start_time:.4f} секунд")
return result
return wrapper

@timing_decorator
def slow_function(delay):
time.sleep(delay)
return "Функция выполнена"

print(slow_function(1))
# Вывод: "Функция slow_function выполнилась за 1.0013 секунд"
# Вывод: "Функция выполнена"

def log_class_usage(cls):
original_methods = {}

# Сохраняем оригинальные методы
for name, method in cls.__dict__.items():
if callable(method) and not name.startswith('__'):
original_methods[name] = method

# Определяем декорированные методы
def create_logged_method(name, method):
def logged_method(self, *args, **kwargs):
print(f"Вызов метода {cls.__name__}.{name}")
return method(self, *args, **kwargs)
return logged_method

# Заменяем методы декорированными версиями
for name, method in original_methods.items():
setattr(cls, name, create_logged_method(name, method))

return cls

@log_class_usage
class Calculator:
def add(self, a, b):
return a + b

def multiply(self, a, b):
return a * b

calc = Calculator()
print(calc.add(2, 3))
# Вывод: "Вызов метода Calculator.add"
# Вывод: 5

from abc import ABC, abstractmethod

# Компонент
class FileSystemItem(ABC):
@abstractmethod
def get_size(self):
pass

@abstractmethod
def display(self, indent=""):
pass

# Лист
class File(FileSystemItem):
def __init__(self, name, size):
self.name = name
self.size = size

def get_size(self):
return self.size

def display(self, indent=""):
print(f"{indent}- File: {self.name}, Size: {self.size} bytes")

# Композит
class Directory(FileSystemItem):
def __init__(self, name):
self.name = name
self.children = []

def add(self, item):
self.children.append(item)

def get_size(self):
return sum(item.get_size() for item in self.children)

def display(self, indent=""):
print(f"{indent}+ Directory: {self.name} (Size: {self.get_size()} bytes)")
for child in self.children:
child.display(indent + " ")

# Использование
root = Directory("root")
documents = Directory("documents")
pictures = Directory("pictures")

root.add(documents)
root.add(pictures)

documents.add(File("resume.docx", 1024))
documents.add(File("cover_letter.pdf", 2048))
pictures.add(File("vacation.jpg", 3072))

root.display()

# Реализация паттерна Observer
class Subject:
def __init__(self):
self._observers = []
self._state = None

def attach(self, observer):
if observer not in self._observers:
self._observers.append(observer)

def detach(self, observer):
try:
self._observers.remove(observer)
except ValueError:
pass

def notify(self):
for observer in self._observers:
observer.update(self)

@property
def state(self):
return self._state

@state.setter
def state(self, value):
self._state = value
self.notify()

class Observer:
def update(self, subject):
pass

# Конкретная реализация
class ConcreteSubject(Subject):
pass

class ConcreteObserverA(Observer):
def update(self, subject):
print(f"ConcreteObserverA: Реагирует на изменение состояния: {subject.state}")

class ConcreteObserverB(Observer):
def update(self, subject):
print(f"ConcreteObserverB: Реагирует на изменение состояния: {subject.state}")

# Использование
subject = ConcreteSubject()

observer_a = ConcreteObserverA()
observer_b = ConcreteObserverB()

subject.attach(observer_a)
subject.attach(observer_b)

subject.state = 123
# Вывод:
# ConcreteObserverA: Реагирует на изменение состояния: 123
# ConcreteObserverB: Реагирует на изменение состояния: 123

# Реализация паттерна Strategy
from abc import ABC, abstractmethod

# Интерфейс стратегии
class SortStrategy(ABC):
@abstractmethod
def sort(self, data):
pass

# Конкретные стратегии
class QuickSort(SortStrategy):
def sort(self, data):
print("Сортировка с помощью QuickSort")
return sorted(data) # Для примера используем встроенную сортировку

class MergeSort(SortStrategy):
def sort(self, data):
print("Сортировка с помощью MergeSort")
return sorted(data) # Для примера используем встроенную сортировку

class BubbleSort(SortStrategy):
def sort(self, data):
print("Сортировка с помощью BubbleSort")
return sorted(data) # Для примера используем встроенную сортировку

# Контекст
class SortContext:
def __init__(self, strategy=None):
self._strategy = strategy

@property
def strategy(self):
return self._strategy

@strategy.setter
def strategy(self, strategy):
self._strategy = strategy

def sort_data(self, data):
if not self._strategy:
raise ValueError("Стратегия не установлена")
return self._strategy.sort(data)

# Использование
context = SortContext()

data = [3, 1, 4, 1, 5, 9, 2, 6, 5, 3, 5]

# Используем различные стратегии
context.strategy = QuickSort()
result1 = context.sort_data(data)

context.strategy = MergeSort()
result2 = context.sort_data(data)

context.strategy = BubbleSort()
result3 = context.sort_data(data)

# Реализация паттерна Command
from abc import ABC, abstractmethod

# Интерфейс команды
class Command(ABC):
@abstractmethod
def execute(self):
pass

@abstractmethod
def undo(self):
pass

# Получатель команды
class Light:
def __init__(self):
self.is_on = False

def turn_on(self):
self.is_on = True
print("Свет включен")

def turn_off(self):
self.is_on = False
print("Свет выключен")

# Конкретные команды
class LightOnCommand(Command):
def __init__(self, light):
self.light = light

def execute(self):
self.light.turn_on()

def undo(self):
self.light.turn_off()

class LightOffCommand(Command):
def __init__(self, light):
self.light = light

def execute(self):
self.light.turn_off()

def undo(self):
self.light.turn_on()

# Инвокер (вызывающий)
class RemoteControl:
def __init__(self):
self.command = None
self.history = []

def set_command(self, command):
self.command = command

def press_button(self):
if self.command:
self.command.execute()
self.history.append(self.command)

def undo_last_action(self):
if self.history:
last_command = self.history.pop()
last_command.undo()

# Использование
light = Light()
light_on = LightOnCommand(light)
light_off = LightOffCommand(light)

remote = RemoteControl()

remote.set_command(light_on)
remote.press_button() # Включаем свет

remote.set_command(light_off)
remote.press_button() # Выключаем свет

remote.undo_last_action() # Отменяем последнее действие (включаем свет)

# Система обработки событий с использованием Observer и Strategy
from abc import ABC, abstractmethod
from enum import Enum, auto
from typing import Dict, List, Any, Type

# Определение типов событий
class EventType(Enum):
USER_CREATED = auto()
USER_UPDATED = auto()
USER_DELETED = auto()
PAYMENT_RECEIVED = auto()
ORDER_PLACED = auto()

# Базовое событие
class Event:
def __init__(self, event_type: EventType, data: Dict[str, Any]):
self.event_type = event_type
self.data = data

# Интерфейс обработчика событий (Strategy)
class EventHandler(ABC):
@abstractmethod
def handle(self, event: Event):
pass

# Конкретные обработчики
class UserCreatedHandler(EventHandler):
def handle(self, event: Event):
print(f"Обработка создания пользователя: {event.data.get('username')}")
# Логика обработки...

class PaymentReceivedHandler(EventHandler):
def handle(self, event: Event):
print(f"Обработка платежа: {event.data.get('amount')} от {event.data.get('user')}")
# Логика обработки...

# Менеджер событий (Observer)
class EventManager:
def __init__(self):
self.handlers: Dict[EventType, List[EventHandler]] = {event_type: [] for event_type in EventType}

def register(self, event_type: EventType, handler: EventHandler):
self.handlers[event_type].append(handler)

def unregister(self, event_type: EventType, handler: EventHandler):
if handler in self.handlers[event_type]:
self.handlers[event_type].remove(handler)

def notify(self, event: Event):
for handler in self.handlers[event.event_type]:
handler.handle(event)

# Пример использования
event_manager = EventManager()

# Регистрация обработчиков
event_manager.register(EventType.USER_CREATED, UserCreatedHandler())
event_manager.register(EventType.PAYMENT_RECEIVED, PaymentReceivedHandler())

# Создание событий
user_event = Event(EventType.USER_CREATED, {"username": "john_doe", "email": "john@example.com"})
payment_event = Event(EventType.PAYMENT_RECEIVED, {"user": "john_doe", "amount": 100.50})

# Обработка событий
event_manager.notify(user_event)
event_manager.notify(payment_event)

# Система плагинов с использованием Factory Method и Strategy
import importlib
import os
from abc import ABC, abstractmethod
from typing import Dict, Type

# Интерфейс плагина (Strategy)
class Plugin(ABC):
@abstractmethod
def process(self, data):
pass

# Фабрика плагинов
class PluginFactory:
_plugins: Dict[str, Type[Plugin]] = {}

@classmethod
def register_plugin(cls, plugin_name: str, plugin_class: Type[Plugin]):
cls._plugins[plugin_name] = plugin_class

@classmethod
def create_plugin(cls, plugin_name: str) -> Plugin:
plugin_class = cls._plugins.get(plugin_name)
if not plugin_class:
raise ValueError(f"Неизвестный плагин: {plugin_name}")
return plugin_class()

@classmethod
def load_plugins(cls, plugins_dir: str):
"""Динамическая загрузка плагинов из указанной директории"""
for filename in os.listdir(plugins_dir):
if filename.endswith(".py") and not filename.startswith("__"):
module_name = filename[:-3] # Удаляем .py
module_path = f"{plugins_dir}.{module_name}"
module = importlib.import_module(module_path)

# Плагины должны регистрироваться при импорте модуля
# Например, через декоратор или вызов метода register_plugin

# Пример регистрации плагина через декоратор
def register(plugin_name: str):
def decorator(plugin_class):
PluginFactory.register_plugin(plugin_name, plugin_class)
return plugin_class
return decorator

# Пример плагина
@register("text_processor")
class TextProcessorPlugin(Plugin):
def process(self, data):
return data.upper() # Простой пример: преобразование текста в верхний регистр

# Использование системы плагинов
PluginFactory.load_plugins("plugins") # Загрузка всех плагинов из директории

# Создание и использование конкретного плагина
plugin = PluginFactory.create_plugin("text_processor")
result = plugin.process("Hello, world!")
print(result) # Вывод: HELLO, WORLD!

# Конфигурационная система с поддержкой различных форматов
import json
import yaml
import configparser
from abc import ABC, abstractmethod
from typing import Dict, Any, Optional

# Интерфейс для чтения конфигурации
class ConfigReader(ABC):
@abstractmethod
def read(self, file_path: str) -> Dict[str, Any]:
pass

# Конкретные реализации для разных форматов
class JSONConfigReader(ConfigReader):
def read(self, file_path: str) -> Dict[str, Any]:
with open(file_path, 'r') as f:
return json.load(f)

class YAMLConfigReader(ConfigReader):
def read(self, file_path: str) -> Dict[str, Any]:
with open(file_path, 'r') as f:
return yaml.safe_load(f)

class INIConfigReader(ConfigReader):
def read(self, file_path: str) -> Dict[str, Any]:
config = configparser.ConfigParser()
config.read(file_path)
# Преобразуем в словарь
return {section: dict(config[section]) for section in config.sections()}

# Адаптер для унифицированного доступа
class ConfigurationManager:
_instance = None

def __new__(cls):
if cls._instance is None:
cls._instance = super(ConfigurationManager, cls).__new__(cls)
cls._instance._readers = {
'.json': JSONConfigReader(),
'.yaml': YAMLConfigReader(),
'.yml': YAMLConfigReader(),
'.ini': INIConfigReader()
}
cls._instance._config = {}
return cls._instance

def load_configuration(self, file_path: str):
"""Загружает конфигурацию из файла, определяя формат по расширению"""
extension = file_path[file_path.rfind('.'):].lower()
reader = self._readers.get(extension)

if not reader:
raise ValueError(f"Неподдерживаемый формат конфигурации: {extension}")

self._config.update(reader.read(file_path))

def get(self, key: str, default: Any = None) -> Any:
"""Получает значение из конфигурации по ключу"""
return self._config.get(key, default)

def get_section(self, section: str) -> Dict[str, Any]:
"""Получает раздел конфигурации"""
return self._config.get(section, {})

# Использование
config = ConfigurationManager()
config.load_configuration("config.json")
config.load_configuration("additional_settings.yaml")

database_url = config.get("database_url", "sqlite:///default.db")
debug_mode = config.get("debug", False)