Паттерны проектирования в современном Python (с примерами из Django/FastAPI)

Введение

Паттерны проектирования — это проверенные временем решения типичных архитектурных проблем. Они не являются готовым кодом, а скорее концептуальными шаблонами, которые можно адаптировать под конкретные задачи. В современной Python-разработке, особенно в контексте таких фреймворков, как Django и FastAPI, понимание паттернов помогает создавать чистый, поддерживаемый и масштабируемый код. В этом посте мы рассмотрим ключевые паттерны, их практическую реализацию на Python и то, как они уже используются в популярных фреймворках.


Паттерны создания (Creational Patterns)

1.1. Фабричный метод (Factory Method) и Абстрактная фабрика (Abstract Factory)

Суть: Делегирование создания объектов специальным методам или классам, чтобы избежать жесткого связывания с конкретными классами.

Пример в FastAPI (Зависимости как фабрика):

from fastapi import Depends, FastAPI
from sqlalchemy.orm import Session

# Абстракция - функция, которая создает и возвращает сессию БД
def get_db_session() -> Session:
    """Фабричный метод для создания сессии БД."""
    db = SessionLocal()
    try:
        yield db  # FastAPI использует это как фабрику зависимостей
    finally:
        db.close()

app = FastAPI()

# Использование: FastAPI автоматически вызывает get_db_session() при запросе
@app.get("/users/{user_id}")
def read_user(user_id: int, db: Session = Depends(get_db_session)):
    # `db` - готовый объект сессии, созданный фабричным методом
    user = db.query(User).filter(User.id == user_id).first()
    return user

Что дает: Управление жизненным циклом объекта (сессии БД) централизовано. Тестирование упрощается (можно подменить get_db_session на мок).

Абстрактная фабрика в Django (Фабрика форм/сериализаторов):

# Упрощенная аналогия: Django REST Framework (DRF) использует фабричный подход
from rest_framework import serializers
from .models import Article, Comment

# Абстрактная "фабрика" serializers.ModelSerializer создает конкретные классы-сериализаторы
class ArticleSerializer(serializers.ModelSerializer):
    class Meta:
        model = Article
        fields = ['id', 'title', 'content']
    # DRF под капотом анализирует модель Article и "создает" поля сериализатора

# В представлении (View) используется паттерн "Фабричный метод"
from rest_framework.generics import CreateAPIView

class CommentCreateView(CreateAPIView):
    serializer_class = CommentSerializer  # Указываем класс, который будет "производить" валидированные объекты
    # CreateAPIView сам вызывает serializer.save() - аналог factory.create()

1.2. Строитель (Builder)

Суть: Пошаговое создание сложного объекта, отделяя конструирование от представления.

Пример: Построение сложного запроса в Django ORM.

class QueryBuilder:
    """Строитель для сложных запросов."""
    def __init__(self, model):
        self.model = model
        self._filters = {}
        self._ordering = []
        self._select_related = []
        self._annotations = {}

    def filter_by_status(self, status: str):
        self._filters['status'] = status
        return self  # Возврат self для цепочки вызовов

    def filter_by_date_range(self, start_date, end_date):
        self._filters['created_at__range'] = (start_date, end_date)
        return self

    def order_by_priority(self):
        self._ordering.append('-priority')
        return self

    def with_related(self, *fields):
        self._select_related.extend(fields)
        return self

    def add_annotation_count(self, field_name):
        from django.db.models import Count
        self._annotations[f'{field_name}_count'] = Count(field_name)
        return self

    def build(self):
        """Финальный метод, возвращающий готовый QuerySet."""
        queryset = self.model.objects.all()
        if self._filters:
            queryset = queryset.filter(**self._filters)
        if self._annotations:
            queryset = queryset.annotate(**self._annotations)
        if self._select_related:
            queryset = queryset.select_related(*self._select_related)
        if self._ordering:
            queryset = queryset.order_by(*self._ordering)
        return queryset

# Использование
from tasks.models import Task

complex_query = (QueryBuilder(Task)
                 .filter_by_status('active')
                 .filter_by_date_range('2023-01-01', '2023-12-31')
                 .with_related('author', 'project')
                 .add_annotation_count('subtasks')
                 .order_by_priority()
                 .build())  # Получаем готовый QuerySet

1.3. Синглтон (Singleton)

Суть: Гарантирует, что у класса существует только один экземпляр, и предоставляет к нему глобальную точку доступа.

Пример в Django: Кеширование и подключения.

# Модуль в Django часто действует как естественный синглтон.
# Явная реализация для менеджера внешних API-клиентов:
from django.conf import settings
import redis
import requests

class ExternalAPIClient:
    _instance = None
    _redis_client = None

    def __new__(cls):
        if cls._instance is None:
            cls._instance = super().__new__(cls)
            cls._instance._initialize()
        return cls._instance

    def _initialize(self):
        """Инициализация тяжелых подключений один раз."""
        self._redis_client = redis.Redis(
            host=settings.REDIS_HOST, port=settings.REDIS_PORT, decode_responses=True
        )
        self._session = requests.Session()
        # Настройка сессии (retry, timeout, headers)

    def get_cached_data(self, key):
        return self._redis_client.get(key)

    def fetch_from_api(self, url):
        # Использует общую сессию для всех вызовов
        return self._session.get(url).json()

# В любом месте проекта получаем один и тот же экземпляр
client = ExternalAPIClient()
data = client.get_cached_data('some_key')

Важно: Вместо собственной реализации часто лучше использовать модуль как синглтон или dependency injection (как в FastAPI).

Паттерны структуры (Structural Patterns)

2.1. Адаптер (Adapter)

Суть: Преобразует интерфейс одного класса в интерфейс, ожидаемый клиентом.

Пример: Адаптер для разных сервисов хранения файлов (Django Storage).

from abc import ABC, abstractmethod
import boto3
from google.cloud import storage as gcs
import os

# Целевой интерфейс, который ожидает наше приложение
class FileStorage(ABC):
    @abstractmethod
    def upload(self, file_path: str, object_name: str) -> str: ...
    @abstractmethod
    def download(self, object_name: str, local_path: str): ...
    @abstractmethod
    def get_url(self, object_name: str) -> str: ...

# Адаптер для AWS S3
class S3StorageAdapter(FileStorage):
    def __init__(self, bucket_name: str):
        self.client = boto3.client('s3')
        self.bucket = bucket_name

    def upload(self, file_path: str, object_name: str) -> str:
        self.client.upload_file(file_path, self.bucket, object_name)
        return f"s3://{self.bucket}/{object_name}"

    def download(self, object_name: str, local_path: str):
        self.client.download_file(self.bucket, object_name, local_path)

    def get_url(self, object_name: str) -> str:
        return f"https://{self.bucket}.s3.amazonaws.com/{object_name}"

# Адаптер для Google Cloud Storage
class GCSStorageAdapter(FileStorage):
    def __init__(self, bucket_name: str):
        self.client = gcs.Client()
        self.bucket = self.client.bucket(bucket_name)

    def upload(self, file_path: str, object_name: str) -> str:
        blob = self.bucket.blob(object_name)
        blob.upload_from_filename(file_path)
        return f"gs://{self.bucket.name}/{object_name}"

# В представлении Django используем единый интерфейс
def upload_profile_avatar(request, storage: FileStorage):
    file = request.FILES['avatar']
    temp_path = f"/tmp/{file.name}"
    with open(temp_path, 'wb') as f:
        for chunk in file.chunks():
            f.write(chunk)

    # Код не знает, S3 это или GCS
    url = storage.upload(temp_path, f"avatars/{request.user.id}")
    os.remove(temp_path)
    return url

# Настройка адаптера через конфиг
if settings.FILE_STORAGE == 's3':
    storage = S3StorageAdapter(settings.AWS_BUCKET)
elif settings.FILE_STORAGE == 'gcs':
    storage = GCSStorageAdapter(settings.GCS_BUCKET)

2.2. Декоратор (Decorator)

Суть: Динамически добавляет объекту новую функциональность, оборачивая его.

Пример в FastAPI: Декораторы эндпоинтов и мидлварей.

from fastapi import FastAPI, HTTPException, Request
import time
from functools import wraps

# 1. Кастомный декоратор для логирования
def log_execution_time(route_function):
    @wraps(route_function)
    async def wrapper(*args, **kwargs):
        start_time = time.time()
        result = await route_function(*args, **kwargs)
        duration = time.time() - start_time
        print(f"{route_function.__name__} выполнено за {duration:.3f}с")
        return result
    return wrapper

# 2. Декоратор для проверки прав
def require_role(required_role: str):
    def decorator(route_function):
        @wraps(route_function)
        async def wrapper(request: Request, *args, **kwargs):
            user_role = request.state.user.get('role')  # Предположим, user добавлен в мидлваре
            if user_role != required_role:
                raise HTTPException(status_code=403, detail="Insufficient permissions")
            return await route_function(request, *args, **kwargs)
        return wrapper
    return decorator

app = FastAPI()

@app.get("/admin/dashboard")
@log_execution_time
@require_role("admin")  # Применяем два декоратора
async def admin_dashboard(request: Request):
    return {"message": "Welcome, Admin"}

# Сам FastAPI построен на идее декораторов: @app.get, @app.post

2.3. Фасад (Facade)

Суть: Предоставляет простой интерфейс к сложной подсистеме.

Пример в Django: Сервисный слой для регистрации пользователя.

# Сложная подсистема
from django.contrib.auth.models import User
from django.core.mail import send_mail
from .models import UserProfile, ActivityLog
from .tasks import send_welcome_email_task
import logging

logger = logging.getLogger(__name__)

# Фасад, скрывающий сложность
class UserRegistrationService:
    """Упрощенный интерфейс для регистрации пользователя."""
    
    @classmethod
    def register_user(cls, username: str, email: str, password: str, **extra_fields):
        # 1. Создание пользователя в Django Auth
        user = User.objects.create_user(username=username, email=email, password=password)
        
        # 2. Создание расширенного профиля
        profile = UserProfile.objects.create(user=user, **extra_fields)
        
        # 3. Логирование события
        ActivityLog.objects.create(user=user, action='registration')
        
        # 4. Асинхронная отправка email через Celery
        send_welcome_email_task.delay(user.id)
        
        # 5. Локальное логирование
        logger.info(f"New user registered: {username} ({email})")
        
        # 6. Возможна интеграция с внешним CRM
        # cls._sync_with_crm(user)
        
        return user

# В представлении — один простой вызов
from django.views import View
from django.http import JsonResponse

class RegisterView(View):
    def post(self, request):
        data = request.POST
        try:
            user = UserRegistrationService.register_user(
                username=data['username'],
                email=data['email'],
                password=data['password'],
                phone_number=data.get('phone')
            )
            return JsonResponse({"status": "success", "user_id": user.id})
        except Exception as e:
            return JsonResponse({"status": "error", "message": str(e)}, status=400)

Паттерны поведения (Behavioral Patterns)

3.1. Стратегия (Strategy)

Суть: Определяет семейство алгоритмов, инкапсулирует каждый и делает их взаимозаменяемыми.

Пример: Разные способы оплаты в Django-приложении.

from abc import ABC, abstractmethod
from decimal import Decimal
import stripe
from paypalcheckoutsdk.orders import OrdersCreateRequest

# Интерфейс стратегии
class PaymentStrategy(ABC):
    @abstractmethod
    def process(self, amount: Decimal, order_id: int, **kwargs) -> dict: ...

# Конкретные стратегии
class StripePaymentStrategy(PaymentStrategy):
    def process(self, amount: Decimal, order_id: int, **kwargs) -> dict:
        token = kwargs['stripe_token']
        # Логика Stripe
        charge = stripe.Charge.create(
            amount=int(amount * 100),  # центы
            currency="usd",
            source=token,
            description=f"Order #{order_id}"
        )
        return {"status": "success", "transaction_id": charge.id}

class PayPalPaymentStrategy(PaymentStrategy):
    def process(self, amount: Decimal, order_id: int, **kwargs) -> dict:
        # Логика PayPal
        request = OrdersCreateRequest()
        request.prefer('return=representation')
        request.request_body({
            "intent": "CAPTURE",
            "purchase_units": [{
                "amount": {"currency_code": "USD", "value": str(amount)}
            }]
        })
        # ... выполнение запроса
        return {"status": "success", "paypal_order_id": "example_id"}

class BankTransferStrategy(PaymentStrategy):
    def process(self, amount: Decimal, order_id: int, **kwargs) -> dict:
        # Генерация реквизитов для перевода
        return {
            "status": "pending",
            "message": "Use these bank details for transfer",
            "details": {"account": "XXX", "amount": amount}
        }

# Контекст, использующий стратегию
class PaymentProcessor:
    def __init__(self, strategy: PaymentStrategy):
        self._strategy = strategy

    def set_strategy(self, strategy: PaymentStrategy):
        self._strategy = strategy

    def execute_payment(self, amount: Decimal, order_id: int, **kwargs):
        return self._strategy.process(amount, order_id, **kwargs)

# Использование в Django View
def create_payment(request, order_id):
    payment_method = request.POST.get('method')
    
    # Выбор стратегии на лету
    if payment_method == 'stripe':
        strategy = StripePaymentStrategy()
    elif payment_method == 'paypal':
        strategy = PayPalPaymentStrategy()
    elif payment_method == 'bank':
        strategy = BankTransferStrategy()
    else:
        return JsonResponse({"error": "Invalid method"}, status=400)
    
    processor = PaymentProcessor(strategy)
    amount = Decimal(request.POST.get('amount'))
    result = processor.execute_payment(amount, order_id, **request.POST.dict())
    
    return JsonResponse(result)

3.2. Наблюдатель (Observer) / Сигналы в Django

Суть: Объект (субъект) уведомляет список наблюдателей об изменении своего состояния.

Пример: Django Signals — встроенная реализация паттерна.

from django.db.models.signals import post_save, pre_delete
from django.dispatch import receiver
from django.contrib.auth.models import User
from .models import UserProfile, AuditLog

# Наблюдатель 1: Автоматическое создание профиля при создании пользователя
@receiver(post_save, sender=User)
def create_user_profile(sender, instance, created, **kwargs):
    if created:
        UserProfile.objects.create(user=instance)
        print(f"Профиль создан для {instance.username}")

# Наблюдатель 2: Логирование удаления
@receiver(pre_delete, sender=User)
def log_user_deletion(sender, instance, **kwargs):
    AuditLog.objects.create(
        action=f"User {instance.username} deleted",
        user_id=instance.id
    )
    print(f"Пользователь {instance.username} будет удален")

# Наблюдатель 3: Отправка уведомления (интеграция с Celery)
@receiver(post_save, sender=User)
def send_welcome_notification(sender, instance, created, **kwargs):
    if created:
        from .tasks import send_welcome_email
        send_welcome_email.delay(instance.email)

Как это работает: Модель User (субъект) отправляет сигналы при сохранении/удалении. Декоратор @receiver регистрирует функции-наблюдатели, которые реагируют на эти события.

3.3. Посредник (Mediator)

Суть: Убирает прямые связи между компонентами, заставляя их общаться через центральный объект-посредник.

Пример: Чат-комната в веб-сокетах (FastAPI/WebSockets).

# Упрощенный пример
from typing import Dict, Set
import asyncio

class ChatMediator:
    """Посредник для управления подключениями чата."""
    def __init__(self):
        self.connections: Dict[str, WebSocket] = {}
        self.rooms: Dict[str, Set[str]] = {"general": set()}

    async def connect(self, user_id: str, websocket: WebSocket):
        await websocket.accept()
        self.connections[user_id] = websocket
        await self.join_room(user_id, "general")

    async def disconnect(self, user_id: str):
        for room in self.rooms.values():
            room.discard(user_id)
        self.connections.pop(user_id, None)

    async def join_room(self, user_id: str, room_name: str):
        if room_name not in self.rooms:
            self.rooms[room_name] = set()
        self.rooms[room_name].add(user_id)
        await self._send_to_user(user_id, f"Joined room {room_name}")

    async def send_message(self, from_user: str, room_name: str, message: str):
        if room_name not in self.rooms:
            return
        for user_id in self.rooms[room_name]:
            if user_id != from_user:  # Не отправляем отправителю
                await self._send_to_user(user_id, f"{from_user}: {message}")

    async def _send_to_user(self, user_id: str, message: str):
        if conn := self.connections.get(user_id):
            try:
                await conn.send_text(message)
            except Exception:
                await self.disconnect(user_id)

# Использование в FastAPI
from fastapi import WebSocket
mediator = ChatMediator()

@app.websocket("/ws/{user_id}")
async def websocket_endpoint(websocket: WebSocket, user_id: str):
    await mediator.connect(user_id, websocket)
    try:
        while True:
            data = await websocket.receive_json()
            # Все сообщения идут через посредника
            await mediator.send_message(
                from_user=user_id,
                room_name=data["room"],
                message=data["text"]
            )
    except Exception:
        await mediator.disconnect(user_id)

Вывод

Паттерны проектирования в Python — это не абстрактная теория, а практические инструменты, которые уже активно используются в экосистеме фреймворков. Django неявно применяет Фабричный метод (менеджеры моделей, фабрики форм), Наблюдателя (сигналы) и Фасад (высокоуровневый ORM API). FastAPI построен на идеях Декоратора (маршрутизация) и Зависимостей (по сути, стратегия предоставления ресурсов).

Главный вывод для современного Python-разработчика: не нужно изобретать сложные реализации паттернов «в лоб». Часто фреймворк уже предоставляет более элегантные встроенные механизмы (как сигналы в Django). Используйте явные паттерны там, где они добавляют ясность и гибкость (Стратегия для алгоритмов, Адаптер для интеграций), но избегайте излишнего усложнения там, где можно обойтись простой функцией или классом. Цель — не «впихнуть» все паттерны в проект, а выбрать те, которые сделают код чище, тестируемее и легче для понимания.