Как применять паттерны программирования в Python: полное руководство

# Классическая реализация Singleton через метакласс
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_string = "postgresql://user:password@localhost/dbname"
print("Database connection initialized")

def connect(self):
print(f"Connecting to {self.connection_string}")
# Логика подключения...

# Демонстрация работы
db1 = DatabaseConnection()
db2 = DatabaseConnection() # Новый экземпляр не создается
print(db1 is db2) # True

def singleton(cls):
instances = {}

def get_instance(*args, **kwargs):
if cls not in instances:
instances[cls] = cls(*args, **kwargs)
return instances[cls]

return get_instance

@singleton
class Logger:
def log(self, msg):
print(f"LOG: {msg}")

from abc import ABC, abstractmethod

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

# Конкретные продукты
class PDFReport(Report):
def generate(self):
return "Generating PDF report..."

class HTMLReport(Report):
def generate(self):
return "Generating HTML report..."

# Абстрактная фабрика
class ReportFactory(ABC):
@abstractmethod
def create_report(self) -> Report:
pass

# Конкретные фабрики
class PDFReportFactory(ReportFactory):
def create_report(self) -> Report:
return PDFReport()

class HTMLReportFactory(ReportFactory):
def create_report(self) -> Report:
return HTMLReport()

# Использование
def client_code(factory: ReportFactory):
report = factory.create_report()
print(report.generate())

# Выбор фабрики может быть на основе конфигурации
factory = PDFReportFactory()
client_code(factory) # "Generating PDF report..."

class Computer:
def __init__(self):
self.cpu = None
self.ram = None
self.storage = None
self.gpu = None

def __str__(self):
return f"Computer: CPU={self.cpu}, RAM={self.ram}GB, Storage={self.storage}GB, GPU={self.gpu}"

class ComputerBuilder:
def __init__(self):
self.computer = Computer()

def configure_cpu(self, cpu):
self.computer.cpu = cpu
return self

def configure_ram(self, ram):
self.computer.ram = ram
return self

def configure_storage(self, storage):
self.computer.storage = storage
return self

def configure_gpu(self, gpu):
self.computer.gpu = gpu
return self

def build(self):
return self.computer

# Использование паттерна Builder с цепочкой методов
gaming_pc = ComputerBuilder() \
.configure_cpu("Intel i9") \
.configure_ram(32) \
.configure_storage(1000) \
.configure_gpu("RTX 3080") \
.build()

print(gaming_pc) # Computer: CPU=Intel i9, RAM=32GB, Storage=1000GB, GPU=RTX 3080

# Существующий класс с несовместимым интерфейсом
class OldSystem:
def specific_request(self):
return "Специфический формат данных старой системы"

# Целевой интерфейс, ожидаемый клиентом
class TargetInterface:
def request(self):
pass

# Адаптер, преобразующий интерфейс OldSystem в TargetInterface
class Adapter(TargetInterface):
def __init__(self, old_system):
self.old_system = old_system

def request(self):
# Преобразование вызова request() в specific_request()
data = self.old_system.specific_request()
return f"Адаптер: (ПРЕОБРАЗОВАНО) {data}"

# Клиентский код, ожидающий объект с интерфейсом TargetInterface
def client_code(target):
print(target.request())

# Использование
old = OldSystem()
adapter = Adapter(old)
client_code(adapter) # Выведет: "Адаптер: (ПРЕОБРАЗОВАНО) Специфический формат данных старой системы"

# Базовый компонент с основной функциональностью
class Component:
def operation(self):
return "Базовая операция"

# Базовый декоратор
class Decorator(Component):
def __init__(self, component):
self._component = component

def operation(self):
return self._component.operation()

# Конкретные декораторы, добавляющие новое поведение
class ConcreteDecoratorA(Decorator):
def operation(self):
return f"ДекораторA({super().operation()})"

class ConcreteDecoratorB(Decorator):
def operation(self):
return f"ДекораторB({super().operation()})"

# Использование
simple = Component()
decorated1 = ConcreteDecoratorA(simple)
decorated2 = ConcreteDecoratorB(decorated1)

print(decorated2.operation()) # Выведет: "ДекораторB(ДекораторA(Базовая операция))"

def timing_decorator(func):
def wrapper(*args, **kwargs):
import time
start = time.time()
result = func(*args, **kwargs)
end = time.time()
print(f"{func.__name__} выполнялась {end – start:.2f} секунд")
return result
return wrapper

@timing_decorator
def slow_function():
import time
time.sleep(1)
return "Функция завершена"

print(slow_function()) # Выведет время выполнения и "Функция завершена"

# Сложная подсистема с множеством классов
class CPU:
def freeze(self):
return "Процессор заморожен"

def jump(self, position):
return f"Процессор прыгнул на позицию {position}"

def execute(self):
return "Процессор выполнил инструкции"

class Memory:
def load(self, position, data):
return f"Память загрузила '{data}' в позицию {position}"

class HardDrive:
def read(self, sector, size):
return f"Жесткий диск прочитал {size} байт из сектора {sector}"

# Фасад, упрощающий взаимодействие с подсистемой
class ComputerFacade:
def __init__(self):
self.cpu = CPU()
self.memory = Memory()
self.hard_drive = HardDrive()

def start(self):
results = []
results.append(self.cpu.freeze())
results.append(self.memory.load(0, "Загрузочный код"))
results.append(self.cpu.jump(0))
results.append(self.cpu.execute())
return "\n".join(results)

# Клиентский код теперь прост
computer = ComputerFacade()
print(computer.start())

# Интерфейс стратегии
from abc import ABC, abstractmethod

class PaymentStrategy(ABC):
@abstractmethod
def pay(self, amount):
pass

# Конкретные стратегии
class CreditCardPayment(PaymentStrategy):
def __init__(self, card_number, cvv, expiration_date):
self.card_number = card_number
self.cvv = cvv
self.expiration_date = expiration_date

def pay(self, amount):
print(f"Оплата {amount} руб. кредитной картой {self.card_number[:4]}****")
return True

class PayPalPayment(PaymentStrategy):
def __init__(self, email, password):
self.email = email
self.password = password

def pay(self, amount):
print(f"Оплата {amount} руб. через PayPal аккаунт {self.email}")
return True

class BitcoinPayment(PaymentStrategy):
def __init__(self, wallet_address):
self.wallet_address = wallet_address

def pay(self, amount):
print(f"Оплата эквивалента {amount} руб. биткоинами на кошелек {self.wallet_address}")
return True

# Контекст, использующий стратегию
class ShoppingCart:
def __init__(self):
self.items = []
self.payment_strategy = None

def add_item(self, item, price):
self.items.append((item, price))

def set_payment_strategy(self, payment_strategy):
self.payment_strategy = payment_strategy

def checkout(self):
if not self.payment_strategy:
raise Exception("Способ оплаты не выбран")

total = sum(price for _, price in self.items)
return self.payment_strategy.pay(total)

# Использование
cart = ShoppingCart()
cart.add_item("Ноутбук", 65000)
cart.add_item("Мышь", 2000)

# Выбор способа оплаты в зависимости от предпочтений пользователя
cart.set_payment_strategy(CreditCardPayment("1234567890123456", "123", "12/25"))
cart.checkout() # Оплата 67000 руб. кредитной картой 1234****

# Смена стратегии
cart.set_payment_strategy(PayPalPayment("user@example.com", "password"))
cart.checkout() # Оплата 67000 руб. через PayPal аккаунт user@example.com

def credit_card_payment(amount, card_details):
print(f"Оплата {amount} руб. кредитной картой {card_details['number'][:4]}****")
return True

def paypal_payment(amount, paypal_details):
print(f"Оплата {amount} руб. через PayPal аккаунт {paypal_details['email']}")
return True

class ShoppingCart:
def __init__(self):
self.items = []
self.payment_strategy = None
self.payment_details = None

def add_item(self, item, price):
self.items.append((item, price))

def set_payment_strategy(self, payment_strategy, payment_details):
self.payment_strategy = payment_strategy
self.payment_details = payment_details

def checkout(self):
total = sum(price for _, price in self.items)
return self.payment_strategy(total, self.payment_details)

# Использование
cart = ShoppingCart()
cart.add_item("Клавиатура", 3000)
cart.set_payment_strategy(
credit_card_payment, 
{"number": "1234567890123456", "cvv": "123", "expiration": "12/25"}
)
cart.checkout() # Оплата 3000 руб. кредитной картой 1234****

# Интерфейс наблюдателя
class Observer(ABC):
@abstractmethod
def update(self, temperature, humidity, pressure):
pass

# Конкретные наблюдатели
class CurrentConditionsDisplay(Observer):
def update(self, temperature, humidity, pressure):
print(f"Текущие условия: {temperature}°C, {humidity}% влажности")

class StatisticsDisplay(Observer):
def __init__(self):
self.temperatures = []

def update(self, temperature, humidity, pressure):
self.temperatures.append(temperature)
avg = sum(self.temperatures) / len(self.temperatures)
print(f"Средняя/Мин/Макс температура: {avg:.1f}°C / {min(self.temperatures)}°C / {max(self.temperatures)}°C")

class ForecastDisplay(Observer):
def __init__(self):
self.last_pressure = 0

def update(self, temperature, humidity, pressure):
prediction = "Улучшение погоды" if pressure > self.last_pressure else "Ухудшение погоды"
self.last_pressure = pressure
print(f"Прогноз: {prediction}")

# Субъект (наблюдаемый объект)
class WeatherData:
def __init__(self):
self.observers = []
self.temperature = 0
self.humidity = 0
self.pressure = 0

def register_observer(self, observer):
if observer not in self.observers:
self.observers.append(observer)

def remove_observer(self, observer):
if observer in self.observers:
self.observers.remove(observer)

def notify_observers(self):
for observer in self.observers:
observer.update(self.temperature, self.humidity, self.pressure)

def set_measurements(self, temperature, humidity, pressure):
self.temperature = temperature
self.humidity = humidity
self.pressure = pressure
self.notify_observers()

# Использование
weather_station = WeatherData()
current_display = CurrentConditionsDisplay()
statistics_display = StatisticsDisplay()
forecast_display = ForecastDisplay()

weather_station.register_observer(current_display)
weather_station.register_observer(statistics_display)
weather_station.register_observer(forecast_display)

weather_station.set_measurements(25, 65, 1013) # Первое измерение
weather_station.set_measurements(26, 70, 1014) # Второе измерение

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

@abstractmethod
def undo(self):
pass

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

def turn_on(self):
self.is_on = True
print(f"{self.location} свет включен")

def turn_off(self):
self.is_on = False
print(f"{self.location} свет выключен")

# Конкретные команды
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.commands = {}
self.history = []

def set_command(self, button, command):
self.commands[button] = command

def press_button(self, button):
if button in self.commands:
self.commands[button].execute()
self.history.append(self.commands[button])
else:
print(f"Кнопка {button} не настроена")

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

# Использование
living_room_light = Light("Гостиная")
kitchen_light = Light("Кухня")

living_room_light_on = LightOnCommand(living_room_light)
living_room_light_off = LightOffCommand(living_room_light)
kitchen_light_on = LightOnCommand(kitchen_light)
kitchen_light_off = LightOffCommand(kitchen_light)

remote = RemoteControl()
remote.set_command("living_room_on", living_room_light_on)
remote.set_command("living_room_off", living_room_light_off)
remote.set_command("kitchen_on", kitchen_light_on)
remote.set_command("kitchen_off", kitchen_light_off)

remote.press_button("living_room_on") # Гостиная свет включен
remote.press_button("kitchen_on") # Кухня свет включен
remote.press_undo() # Кухня свет выключен (отмена последней команды)
remote.press_button("living_room_off") # Гостиная свет выключен

class Node:
def __init__(self, value):
self.value = value
self.left = None
self.right = None

class BinaryTree:
def __init__(self, root=None):
self.root = root

def in_order_iterator(self):
def in_order_traversal(node):
if node:
yield from in_order_traversal(node.left)
yield node.value
yield from in_order_traversal(node.right)

yield from in_order_traversal(self.root)

# Реализация протокола итерации
def __iter__(self):
return self.in_order_iterator()

# Создание бинарного дерева
root = Node(10)
root.left = Node(5)
root.right = Node(15)
root.left.left = Node(3)
root.left.right = Node(7)
root.right.left = Node(12)
root.right.right = Node(18)

tree = BinaryTree(root)

# Использование итератора
print("In-order traversal:")
for value in tree:
print(value, end=" ") # 3 5 7 10 12 15 18

# Пример фабричного метода в Flask
def create_app(config_name='default'):
app = Flask(__name__)

# Загрузка конфигурации в зависимости от окружения
if config_name == 'development':
app.config.from_object('config.DevelopmentConfig')
elif config_name == 'testing':
app.config.from_object('config.TestingConfig')
else:
app.config.from_object('config.ProductionConfig')

# Регистрация расширений и маршрутов
db.init_app(app)
register_blueprints(app)

return app

# Пример паттерна Repository
class UserRepository:
def __init__(self, db_session):
self.session = db_session

def get_by_id(self, user_id):
return self.session.query(User).filter(User.id == user_id).first()

def get_all(self):
return self.session.query(User).all()

def add(self, user):
self.session.add(user)
self.session.commit()
return user

def update(self, user):
self.session.merge(user)
self.session.commit()
return user

def delete(self, user_id):
user = self.get_by_id(user_id)
if user:
self.session.delete(user)
self.session.commit()
return True
return False

# Использование в сервисном слое
class UserService:
def __init__(self, user_repository):
self.repository = user_repository

def register_user(self, username, email, password):
# Бизнес-логика
if self.repository.get_by_email(email):
raise ValueError("User with this email already exists")

password_hash = hash_password(password)
user = User(username=username, email=email, password=password_hash)
return self.repository.add(user)

import asyncio

# Пример Producer-Consumer с asyncio
async def producer(queue, n):
for i in range(n):
# Производим данные
item = f"Item {i}"

# Кладем в очередь
await queue.put(item)
print(f"Produced {item}")

# Симуляция работы
await asyncio.sleep(0.1)

# Сигнал о завершении
await queue.put(None)

async def consumer(queue):
while True:
# Получаем данные из очереди
item = await queue.get()

# Проверяем сигнал о завершении
if item is None:
break

# Обрабатываем данные
print(f"Consumed {item}")

# Симуляция работы
await asyncio.sleep(0.2)

# Отмечаем задачу как выполненную
queue.task_done()

async def main():
queue = asyncio.Queue(maxsize=5) # Ограниченная очередь

# Запускаем producer и consumer
producer_task = asyncio.create_task(producer(queue, 10))
consumer_task = asyncio.create_task(consumer(queue))

# Ждем завершения всех задач
await producer_task
await consumer_task

# Запуск
asyncio.run(main())

# До рефакторинга: прямое создание объектов и жесткие зависимости
def process_user_data(data):
# Напрямую создаем объекты и обрабатываем данные
user = User(data['name'], data['email'])
db = Database()
db.save(user)

# Отправляем уведомление по email
mailer = EmailService()
mailer.send_welcome_email(user.email)

return user.id

# После рефакторинга: использование DI и Factory
class UserProcessor:
def __init__(self, user_factory, storage_service, notification_service):
self.user_factory = user_factory
self.storage = storage_service
self.notifier = notification_service

def process(self, data):
# Создаем пользователя через фабрику
user = self.user_factory.create_user(data)

# Сохраняем через абстракцию хранилища
self.storage.save(user)

# Уведомляем через абстракцию сервиса уведомлений
self.notifier.notify_user_created(user)

return user.id

# Использование
processor = UserProcessor(
UserFactory(), 
DatabaseStorage(db_connection), 
EmailNotificationService(email_config)
)
user_id = processor.process(user_data)