Шаблоны проектирования Python для чистой архитектуры

Создавайте поддерживаемые приложения на Python с помощью принципов SOLID

Чистая архитектура революционизировала подход разработчиков к созданию масштабируемых, поддерживаемых приложений, делая акцент на разделение ответственности и управлении зависимостями.

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

Яркий зал на технологической конференции

Понимание чистой архитектуры в Python

Чистая архитектура, предложенная Робертом К. Мартином (Дядя Боб), организует программное обеспечение в концентрические слои, где зависимости направлены внутрь к ядру бизнес-логики. Этот архитектурный паттерн гарантирует, что критически важные бизнес-правила вашего приложения остаются независимыми от фреймворков, баз данных и внешних сервисов.

Основная философия

Основной принцип прост, но мощен: бизнес-логика не должна зависеть от инфраструктуры. Ваши доменные сущности, кейсы использования и бизнес-правила должны работать независимо от того, используете ли вы PostgreSQL или MongoDB, FastAPI или Flask, AWS или Azure.

В Python эта философия идеально сочетается с “утиной типизацией” и ориентированным на протоколы программированием языка, позволяя чистое разделение без формальностей, необходимых в статически типизированных языках.

Четыре слоя чистой архитектуры

Слой сущностей (Домен): Чистые бизнес-объекты с правилами, применимыми на уровне предприятия. Это POJOs (Plain Old Python Objects) без внешних зависимостей.

Слой кейсов использования (Приложение): Бизнес-правила, специфичные для приложения, которые организуют поток данных между сущностями и внешними сервисами.

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

Слой фреймворков и драйверов: Все внешние детали, такие как базы данных, веб-фреймворки и внешние API.

Принципы SOLID в Python

Принципы SOLID составляют основу чистой архитектуры. Давайте рассмотрим, как каждый принцип проявляется в Python. Для всестороннего обзора шаблонов проектирования в Python см. Руководство по шаблонам проектирования в Python.

Принцип единственной ответственности (SRP)

Каждый класс должен иметь одну причину для изменения:

# Плохо: Множество ответственностей
class UserManager:
    def create_user(self, user_data):
        # Создание пользователя
        pass

    def send_welcome_email(self, user):
        # Отправка email
        pass

    def log_creation(self, user):
        # Логирование в файл
        pass

# Хорошо: Разделенные ответственности
class UserService:
    def __init__(self, repository, email_service, logger):
        self.repository = repository
        self.email_service = email_service
        self.logger = logger

    def create_user(self, user_data):
        user = User(**user_data)
        self.repository.save(user)
        self.email_service.send_welcome(user)
        self.logger.info(f"Пользователь создан: {user.id}")
        return user

Принцип открытости/закрытости (OCP)

Программные сущности должны быть открыты для расширения, но закрыты для модификации:

from abc import ABC, abstractmethod
from typing import Protocol

# Использование Protocol (Python 3.8+)
class PaymentProcessor(Protocol):
    def process_payment(self, amount: float) -> bool:
        ...

class CreditCardProcessor:
    def process_payment(self, amount: float) -> bool:
        # Логика кредитной карты
        return True

class PayPalProcessor:
    def process_payment(self, amount: float) -> bool:
        # Логика PayPal
        return True

# Легко расширяется без модификации существующего кода
class CryptoProcessor:
    def process_payment(self, amount: float) -> bool:
        # Логика криптовалюты
        return True

Принцип подстановки Лисков (LSP)

Объекты должны быть заменяемыми их подтипами без нарушения работы программы:

from abc import ABC, abstractmethod

class DataStore(ABC):
    @abstractmethod
    def save(self, key: str, value: str) -> None:
        pass

    @abstractmethod
    def get(self, key: str) -> str:
        pass

class PostgreSQLStore(DataStore):
    def save(self, key: str, value: str) -> None:
        # Реализация PostgreSQL
        pass

    def get(self, key: str) -> str:
        # Реализация PostgreSQL
        return ""

class RedisStore(DataStore):
    def save(self, key: str, value: str) -> None:
        # Реализация Redis
        pass

    def get(self, key: str) -> str:
        # Реализация Redis
        return ""

# Оба могут использоваться взаимозаменяемо
def process_data(store: DataStore, key: str, value: str):
    store.save(key, value)
    return store.get(key)

Принцип разделения интерфейса (ISP)

Клиенты не должны быть вынуждены зависеть от интерфейсов, которые они не используют:

# Плохо: Толстый интерфейс
class Worker(ABC):
    @abstractmethod
    def work(self): pass

    @abstractmethod
    def eat(self): pass

    @abstractmethod
    def sleep(self): pass

# Хорошо: Разделенные интерфейсы
class Workable(Protocol):
    def work(self) -> None: ...

class Eatable(Protocol):
    def eat(self) -> None: ...

class Human:
    def work(self) -> None:
        print("Работаю")

    def eat(self) -> None:
        print("Ем")

class Robot:
    def work(self) -> None:
        print("Работаю")
    # Метод eat не нужен

Принцип инверсии зависимостей (DIP)

Высокоуровневые модули не должны зависеть от низкоуровневых модулей. Оба должны зависеть от абстракций:

from typing import Protocol

# Абстракция
class EmailSender(Protocol):
    def send(self, to: str, subject: str, body: str) -> None:
        ...

# Низкоуровневый модуль
class SMTPEmailSender:
    def send(self, to: str, subject: str, body: str) -> None:
        # Реализация SMTP
        pass

# Высокоуровневый модуль зависит от абстракции
class UserRegistrationService:
    def __init__(self, email_sender: EmailSender):
        self.email_sender = email_sender

    def register(self, email: str, name: str):
        # Логика регистрации
        self.email_sender.send(
            to=email,
            subject="Добро пожаловать!",
            body=f"Привет {name}"
        )

Паттерн Репозиторий: Абстрагирование доступа к данным

Паттерн Репозиторий предоставляет коллекционный интерфейс для доступа к доменным объектам, скрывая детали хранения данных.

Базовая реализация репозитория

from abc import ABC, abstractmethod
from typing import List, Optional
from dataclasses import dataclass
from uuid import UUID, uuid4

@dataclass
class User:
    id: UUID
    email: str
    name: str
    is_active: bool = True

class UserRepository(ABC):
    @abstractmethod
    def save(self, user: User) -> User:
        pass

    @abstractmethod
    def get_by_id(self, user_id: UUID) -> Optional[User]:
        pass

    @abstractmethod
    def get_by_email(self, email: str) -> Optional[User]:
        pass

    @abstractmethod
    def list_all(self) -> List[User]:
        pass

    @abstractmethod
    def delete(self, user_id: UUID) -> bool:
        pass

Реализация с SQLAlchemy

from sqlalchemy import create_engine, Column, String, Boolean
from sqlalchemy.dialects.postgresql import UUID as PGUUID
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, Session

Base = declarative_base()

class UserModel(Base):
    __tablename__ = 'users'

    id = Column(PGUUID(as_uuid=True), primary_key=True)
    email = Column(String, unique=True, nullable=False)
    name = Column(String, nullable=False)
    is_active = Column(Boolean, default=True)

class SQLAlchemyUserRepository(UserRepository):
    def __init__(self, session: Session):
        self.session = session

    def save(self, user: User) -> User:
        user_model = UserModel(
            id=user.id,
            email=user.email,
            name=user.name,
            is_active=user.is_active
        )
        self.session.add(user_model)
        self.session.commit()
        return user

    def get_by_id(self, user_id: UUID) -> Optional[User]:
        user_model = self.session.query(UserModel).filter(
            UserModel.id == user_id
        ).first()

        if not user_model:
            return None

        return User(
            id=user_model.id,
            email=user_model.email,
            name=user_model.name,
            is_active=user_model.is_active
        )

    def get_by_email(self, email: str) -> Optional[User]:
        user_model = self.session.query(UserModel).filter(
            UserModel.email == email
        ).first()

        if not user_model:
            return None

        return User(
            id=user_model.id,
            email=user_model.email,
            name=user_model.name,
            is_active=user_model.is_active
        )

    def list_all(self) -> List[User]:
        users = self.session.query(UserModel).all()
        return [
            User(
                id=u.id,
                email=u.email,
                name=u.name,
                is_active=u.is_active
            )
            for u in users
        ]

    def delete(self, user_id: UUID) -> bool:
        result = self.session.query(UserModel).filter(
            UserModel.id == user_id
        ).delete()
        self.session.commit()
        return result > 0

Репозиторий в памяти для тестирования

class InMemoryUserRepository(UserRepository):
    def __init__(self):
        self.users: dict[UUID, User] = {}

    def save(self, user: User) -> User:
        self.users[user.id] = user
        return user

    def get_by_id(self, user_id: UUID) -> Optional[User]:
        return self.users.get(user_id)

    def get_by_email(self, email: str) -> Optional[User]:
        for user in self.users.values():
            if user.email == email:
                return user
        return None

    def list_all(self) -> List[User]:
        return list(self.users.values())

    def delete(self, user_id: UUID) -> bool:
        if user_id in self.users:
            del self.users[user_id]
            return True
        return False

Сервисный слой: Оркестрация бизнес-логики

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

from typing import Optional
from uuid import uuid4

class UserAlreadyExistsError(Exception):
    pass

class UserNotFoundError(Exception):
    pass

class UserService:
    def __init__(
        self,
        user_repository: UserRepository,
        email_service: EmailSender,
        event_publisher: 'EventPublisher'
    ):
        self.user_repository = user_repository
        self.email_service = email_service
        self.event_publisher = event_publisher

    def register_user(self, email: str, name: str) -> User:
        # Проверка существования пользователя
        existing_user = self.user_repository.get_by_email(email)
        if existing_user:
            raise UserAlreadyExistsError(f"Пользователь с email {email} уже существует")

        # Создание нового пользователя
        user = User(
            id=uuid4(),
            email=email,
            name=name,
            is_active=True
        )

        # Сохранение в репозиторий
        user = self.user_repository.save(user)

        # Отправка приветственного письма
        self.email_service.send(
            to=user.email,
            subject="Добро пожаловать!",
            body=f"Привет {user.name}, добро пожаловать на нашу платформу!"
        )

        # Публикация события
        self.event_publisher.publish('user.registered', {
            'user_id': str(user.id),
            'email': user.email
        })

        return user

    def deactivate_user(self, user_id: UUID) -> User:
        user = self.user_repository.get_by_id(user_id)
        if not user:
            raise UserNotFoundError(f"Пользователь {user_id} не найден")

        user.is_active = False
        user = self.user_repository.save(user)

        self.event_publisher.publish('user.deactivated', {
            'user_id': str(user.id)
        })

        return user

Внедрение зависимостей в Python

Динамическая природа Python делает внедрение зависимостей простым без необходимости использования сложных фреймворков.

Конструкторное внедрение

class OrderService:
    def __init__(
        self,
        order_repository: 'OrderRepository',
        payment_processor: PaymentProcessor,
        notification_service: 'NotificationService'
    ):
        self.order_repository = order_repository
        self.payment_processor = payment_processor
        self.notification_service = notification_service

    def place_order(self, order_data: dict):
        # Использование внедренных зависимостей
        pass

Простой контейнер зависимостей

from typing import Dict, Type, Callable, Any

class Container:
    def __init__(self):
        self._services: Dict[Type, Callable] = {}
        self._singletons: Dict[Type, Any] = {}

    def register(self, interface: Type, factory: Callable):
        self._services[interface] = factory

    def register_singleton(self, interface: Type, instance: Any):
        self._singletons[interface] = instance

    def resolve(self, interface: Type):
        if interface in self._singletons:
            return self._singletons[interface]

        factory = self._services.get(interface)
        if factory:
            return factory(self)

        raise ValueError(f"Регистрация не найдена для {interface}")

# Использование
def create_container() -> Container:
    container = Container()

    # Регистрация сервисов
    container.register_singleton(
        Session,
        sessionmaker(bind=create_engine('postgresql://...'))()
    )

    container.register(
        UserRepository,
        lambda c: SQLAlchemyUserRepository(c.resolve(Session))
    )

    container.register(
        EmailSender,
        lambda c: SMTPEmailSender()
    )

    container.register(
        UserService,
        lambda c: UserService(
            c.resolve(UserRepository),
            c.resolve(EmailSender),
            c.resolve(EventPublisher)
        )
    )

    return container

Гексагональная архитектура (Порты и адаптеры)

Гексагональная архитектура размещает бизнес-логику в центре с адаптерами, обрабатывающими внешнюю коммуникацию.

Определение портов (интерфейсов)

# Входной порт (основной)
class CreateUserUseCase(Protocol):
    def execute(self, request: 'CreateUserRequest') -> 'CreateUserResponse':
        ...

# Выходной порт (вторичный)
class UserPersistencePort(Protocol):
    def save(self, user: User) -> User:
        ...

    def find_by_email(self, email: str) -> Optional[User]:
        ...

Реализация адаптеров

from pydantic import BaseModel, EmailStr

# Входной адаптер (REST API)
from fastapi import FastAPI, Depends, HTTPException

class CreateUserRequest(BaseModel):
    email: EmailStr
    name: str

class CreateUserResponse(BaseModel):
    id: str
    email: str
    name: str

app = FastAPI()

@app.post("/users", response_model=CreateUserResponse)
def create_user(
    request: CreateUserRequest,
    user_service: UserService = Depends(get_user_service)
):
    try:
        user = user_service.register_user(
            email=request.email,
            name=request.name
        )
        return CreateUserResponse(
            id=str(user.id),
            email=user.email,
            name=user.name
        )
    except UserAlreadyExistsError as e:
        raise HTTPException(status_code=400, detail=str(e))

# Выходной адаптер (база данных)
# Уже реализован как SQLAlchemyUserRepository

Паттерны Domain-Driven Design

Value Objects

Неизменяемые объекты, определенные своими атрибутами:

from dataclasses import dataclass
from typing import Pattern
import re

@dataclass(frozen=True)
class Email:
    value: str

    EMAIL_PATTERN: Pattern = re.compile(r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$')

    def __post_init__(self):
        if not self.EMAIL_PATTERN.match(self.value):
            raise ValueError(f"Неверный email: {self.value}")

    def __str__(self):
        return self.value

@dataclass(frozen=True)
class Money:
    amount: float
    currency: str

    def __post_init__(self):
        if self.amount < 0:
            raise ValueError("Сумма не может быть отрицательной")
        if self.currency not in ['USD', 'EUR', 'GBP']:
            raise ValueError(f"Неподдерживаемая валюта: {self.currency}")

    def add(self, other: 'Money') -> 'Money':
        if self.currency != other.currency:
            raise ValueError("Нельзя складывать разные валюты")
        return Money(self.amount + other.amount, self.currency)

Агрегаты

Кластер доменных объектов, рассматриваемых как единое целое:

from dataclasses import dataclass, field
from typing import List
from datetime import datetime

@dataclass
class OrderItem:
    product_id: UUID
    quantity: int
    price: Money

    def total(self) -> Money:
        return Money(
            self.price.amount * self.quantity,
            self.price.currency
        )

@dataclass
class Order:
    id: UUID
    customer_id: UUID
    items: List[OrderItem] = field(default_factory=list)
    status: str = "pending"
    created_at: datetime = field(default_factory=datetime.now)

    def add_item(self, product_id: UUID, quantity: int, price: Money):
        item = OrderItem(product_id, quantity, price)
        self.items.append(item)

    def remove_item(self, product_id: UUID):
        self.items = [
            item for item in self.items
            if item.product_id != product_id
        ]

    def total(self) -> Money:
        if not self.items:
            return Money(0, "USD")

        return sum(
            (item.total() for item in self.items),
            Money(0, self.items[0].price.currency)
        )

    def confirm(self):
        if not self.items:
            raise ValueError("Нельзя подтвердить пустой заказ")
        if self.status != "pending":
            raise ValueError("Заказ уже обработан")
        self.status = "confirmed"

Доменные события

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

from dataclasses import dataclass
from datetime import datetime
from typing import List, Callable

@dataclass
class DomainEvent:
    occurred_at: datetime = field(default_factory=datetime.now)

@dataclass
class OrderConfirmed(DomainEvent):
    order_id: UUID
    customer_id: UUID
    total: Money

class EventPublisher:
    def __init__(self):
        self._handlers: Dict[Type, List[Callable]] = {}

    def subscribe(self, event_type: Type, handler: Callable):
        if event_type not in self._handlers:
            self._handlers[event_type] = []
        self._handlers[event_type].append(handler)

    def publish(self, event: DomainEvent):
        event_type = type(event)
        handlers = self._handlers.get(event_type, [])
        for handler in handlers:
            handler(event)

Современные возможности Python для реализации чистой архитектуры

Python’s modern features make implementing clean architecture more elegant and type-safe. If you need a quick reference for Python syntax and features, check out the Python Cheatsheet.

Типы и Протоколы

from typing import Protocol, runtime_checkable

@runtime_checkable
class Serializable(Protocol):
    def to_dict(self) -> dict:
        ...

    @classmethod
    def from_dict(cls, data: dict) -> 'Serializable':
        ...
def serialize(obj: Serializable) -> dict:
    return obj.to_dict()

Pydantic для валидации

from pydantic import BaseModel, Field, validator
from typing import Optional

class CreateUserDTO(BaseModel):
    email: EmailStr
    name: str = Field(..., min_length=2, max_length=100)
    age: Optional[int] = Field(None, ge=0, le=150)

    @validator('name')
    def name_must_not_contain_numbers(cls, v):
        if any(char.isdigit() for char in v):
            raise ValueError('Name cannot contain numbers')
        return v

    class Config:
        frozen = True  # Make immutable

Async/Await для операций ввода-вывода

Python’s async/await syntax is particularly powerful for I/O-bound operations in clean architecture, allowing non-blocking interactions with databases and external services. When deploying Python applications to serverless platforms, understanding performance characteristics becomes crucial—see AWS lambda performance: JavaScript vs Python vs Golang for insights on optimizing Python serverless functions.

from typing import List
import asyncio

class AsyncUserRepository(ABC):
    @abstractmethod
    async def save(self, user: User) -> User:
        pass

    @abstractmethod
    async def get_by_id(self, user_id: UUID) -> Optional[User]:
        pass

class AsyncUserService:
    def __init__(self, repository: AsyncUserRepository):
        self.repository = repository

    async def register_user(self, email: str, name: str) -> User:
        user = User(id=uuid4(), email=email, name=name)
        return await self.repository.save(user)

    async def get_users_batch(self, user_ids: List[UUID]) -> List[User]:
        tasks = [self.repository.get_by_id(uid) for uid in user_ids]
        results = await asyncio.gather(*tasks)
        return [u for u in results if u is not None]

Лучшие практики структуры проекта

Правильная организация проекта важна для поддержания чистой архитектуры. Перед настройкой структуры проекта убедитесь, что вы используете виртуальные окружения Python для изоляции зависимостей. venv Cheatsheet содержит всю необходимую информацию о управлении виртуальными окружениями. Для современных проектов на Python рассмотрите использование uv - New Python Package, Project, and Environment Manager, который обеспечивает более быстрый менеджмент пакетов и настройку проектов.

my_application/
├── domain/                 # Предприятие бизнес-правила
│   ├── __init__.py
│   ├── entities/
│   │   ├── __init__.py
│   │   ├── user.py
│   │   └── order.py
│   ├── value_objects/
│   │   ├── __init__.py
│   │   ├── email.py
│   │   └── money.py
│   ├── events/
│   │   ├── __init__.py
│   │   └── user_events.py
│   └── exceptions.py
│
├── application/            # Приложение бизнес-правила
│   ├── __init__.py
│   ├── use_cases/
│   │   ├── __init__.py
│   │   ├── create_user.py
│   │   └── place_order.py
│   ├── services/
│   │   ├── __init__.py
│   │   └── user_service.py
│   └── ports/
│       ├── __init__.py
│       ├── repositories.py
│       └── external_services.py
│
├── infrastructure/         # Внешние интерфейсы
│   ├── __init__.py
│   ├── persistence/
│   │   ├── __init__.py
│   │   ├── sqlalchemy/
│   │   │   ├── models.py
│   │   │   └── repositories.py
│   │   └── mongodb/
│   │       └── repositories.py
│   ├── messaging/
│   │   ├── __init__.py
│   │   └── rabbitmq_publisher.py
│   ├── external_services/
│   │   ├── __init__.py
│   │   └── email_service.py
│   └── config.py
│
├── presentation/           # UI/API слой
│   ├── __init__.py
│   ├── api/
│   │   ├── __init__.py
│   │   ├── dependencies.py
│   │   ├── routes/
│   │   │   ├── __init__.py
│   │   │   ├── users.py
│   │   │   └── orders.py
│   │   └── schemas/
│   │       ├── __init__.py
│   │       └── user_schemas.py
│   └── cli/
│       └── commands.py
│
├── tests/
│   ├── unit/
│   ├── integration/
│   └── e2e/
│
├── main.py                 # Точка входа приложения
├── container.py            # Настройка инъекции зависимостей
├── pyproject.toml
└── README.md

Тестирование чистой архитектуры

Юнит-тестирование доменной логики

import pytest
from uuid import uuid4

def test_user_creation():
    user = User(
        id=uuid4(),
        email="test@example.com",
        name="Test User"
    )
    assert user.email == "test@example.com"
    assert user.is_active is True

def test_order_total_calculation():
    order = Order(id=uuid4(), customer_id=uuid4())
    order.add_item(
        uuid4(),
        quantity=2,
        price=Money(10.0, "USD")
    )
    order.add_item(
        uuid4(),
        quantity=1,
        price=Money(5.0, "USD")
    )
    assert order.total().amount == 25.0

Интеграционное тестирование с репозиторием

@pytest.fixture
def in_memory_repository():
    return InMemoryUserRepository()

def test_user_repository_save_and_retrieve(in_memory_repository):
    user = User(
        id=uuid4(),
        email="test@example.com",
        name="Test User"
    )

    saved_user = in_memory_repository.save(user)
    retrieved_user = in_memory_repository.get_by_id(user.id)

    assert retrieved_user is not None
    assert retrieved_user.email == user.email

Тестирование слоя сервисов

from unittest.mock import Mock

def test_user_registration():
    # Arrange
    mock_repository = Mock(spec=UserRepository)
    mock_repository.get_by_email.return_value = None
    mock_repository.save.return_value = User(
        id=uuid4(),
        email="test@example.com",
        name="Test"
    )

    mock_email = Mock(spec=EmailSender)
    mock_events = Mock(spec=EventPublisher)

    service = UserService(mock_repository, mock_email, mock_events)

    # Act
    user = service.register_user("test@example.com", "Test")

    # Assert
    assert user.email == "test@example.com"
    mock_repository.save.assert_called_once()
    mock_email.send.assert_called_once()
    mock_events.publish.assert_called_once()

Общие ошибки и способы их избежания

Переинжиниринг

Не реализуйте чистую архитектуру для простых CRUD-приложений. Начните с простого и рефакторите по мере роста сложности.

Утечки абстракций

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

# Плохо
from sqlalchemy import Column

@dataclass
class User:
    id: Column(Integer, primary_key=True)  # Утечка фреймворка в домен

# Хорошо
@dataclass
class User:
    id: UUID  # Чистая доменная сущность

Циклические зависимости

Используйте инъекцию зависимостей и интерфейсы для устранения циклических зависимостей между слоями.

Игнорирование контекста

Чистая архитектура не является универсальным решением. Настройте строгость слоев в зависимости от размера проекта и опыта команды.

Полезные ссылки