Паттерны проектирования в современном 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). Используйте явные паттерны там, где они добавляют ясность и гибкость (Стратегия для алгоритмов, Адаптер для интеграций), но избегайте излишнего усложнения там, где можно обойтись простой функцией или классом. Цель — не «впихнуть» все паттерны в проект, а выбрать те, которые сделают код чище, тестируемее и легче для понимания.