Что такое гексагональная архитектура?

Гексагональная архитектура

Гексагональная архитектура (Hexagonal Architecture), также известная как Ports and Adapters (Порты и Адаптеры), — это архитектурный паттерн, придуманный Alistair Cockburn для построения систем, которые легко тестировать и поддерживать. Основная идея заключается в отделении бизнес-логики от внешних зависимостей.

Основная концепция

Архитектура состоит из трёх слоёв:

1. Ядро (Domain/Application) — чистая бизнес-логика
2. Порты (Ports) — интерфейсы для взаимодействия
3. Адаптеры (Adapters) — реализации для внешних систем

          ┌─────────────────────┐
          │  Внешние системы    │
          │  (БД, API, UI)      │
          └──────────┬──────────┘
                     │
          ┌──────────▼──────────┐
          │   Адаптеры (Out)    │
          │  (API клиенты,      │
          │   репозитории)      │
          └──────────┬──────────┘
                     │
          ┌──────────▼──────────┐
          │  Порты (Interfaces) │
          └──────────┬──────────┘
                     │
          ┌──────────▼──────────┐
          │   Ядро (Domain)     │
          │  (Use Cases,        │
          │   Entities)         │
          └──────────┬──────────┘
                     │
          ┌──────────▼──────────┐
          │  Порты (Interfaces) │
          └──────────┬──────────┘
                     │
          ┌──────────▼──────────┐
          │  Адаптеры (In)      │
          │  (Controllers,      │
          │   HTTP handlers)    │
          └──────────┬──────────┘
                     │
          ┌──────────▼──────────┐
          │  Пользователи, API  │
          │      клиенты        │
          └─────────────────────┘

Примеры реализации на Python

Шаг 1: Определяем порты (интерфейсы)

from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import Optional, List

# Модель домена (независима от фреймворков)
@dataclass
class User:
    id: int
    username: str
    email: str

# Входящий порт (Input Port) — определяет вход в приложение
class UserUseCasePort(ABC):
    @abstractmethod
    def create_user(self, username: str, email: str) -> User:
        pass
    
    @abstractmethod
    def get_user(self, user_id: int) -> Optional[User]:
        pass
    
    @abstractmethod
    def list_users(self) -> List[User]:
        pass

# Исходящий порт (Output Port) — позволяет ядру обращаться к внешнему миру
class UserRepositoryPort(ABC):
    @abstractmethod
    def save(self, user: User) -> None:
        pass
    
    @abstractmethod
    def find_by_id(self, user_id: int) -> Optional[User]:
        pass
    
    @abstractmethod
    def find_all(self) -> List[User]:
        pass

class EmailServicePort(ABC):
    @abstractmethod
    def send_confirmation(self, email: str) -> bool:
        pass

Шаг 2: Реализуем ядро (Application/Domain)

from typing import Optional, List

class CreateUserUseCase(UserUseCasePort):
    """Реализация бизнес-логики создания пользователя."""
    
    def __init__(
        self,
        user_repository: UserRepositoryPort,
        email_service: EmailServicePort
    ):
        # Инъекция зависимостей через конструктор
        self.user_repository = user_repository
        self.email_service = email_service
    
    def create_user(self, username: str, email: str) -> User:
        """Создаёт пользователя и отправляет письмо подтверждения."""
        # Бизнес-логика (не зависит от реализации БД или Email)
        if not self._validate_email(email):
            raise ValueError(f"Invalid email: {email}")
        
        user = User(
            id=self._generate_id(),
            username=username,
            email=email
        )
        
        # Сохраняем в репозиторий
        self.user_repository.save(user)
        
        # Отправляем письмо
        if not self.email_service.send_confirmation(email):
            raise Exception(f"Failed to send email to {email}")
        
        return user
    
    def get_user(self, user_id: int) -> Optional[User]:
        return self.user_repository.find_by_id(user_id)
    
    def list_users(self) -> List[User]:
        return self.user_repository.find_all()
    
    @staticmethod
    def _validate_email(email: str) -> bool:
        return '@' in email
    
    @staticmethod
    def _generate_id() -> int:
        import uuid
        return int(str(uuid.uuid4().int)[:8])

Шаг 3: Реализуем адаптеры (Adapters)

from typing import Dict, Optional, List

# Адаптер для БД (Out Adapter)
class SQLAlchemyUserRepository(UserRepositoryPort):
    def __init__(self, db_session):
        self.session = db_session
    
    def save(self, user: User) -> None:
        # Сохранение в реальную БД
        db_user = UserModel(
            id=user.id,
            username=user.username,
            email=user.email
        )
        self.session.add(db_user)
        self.session.commit()
    
    def find_by_id(self, user_id: int) -> Optional[User]:
        db_user = self.session.query(UserModel).filter_by(id=user_id).first()
        if db_user:
            return User(
                id=db_user.id,
                username=db_user.username,
                email=db_user.email
            )
        return None
    
    def find_all(self) -> List[User]:
        db_users = self.session.query(UserModel).all()
        return [
            User(id=u.id, username=u.username, email=u.email)
            for u in db_users
        ]

# Адаптер для Email сервиса (Out Adapter)
class SMTPEmailService(EmailServicePort):
    def __init__(self, smtp_host: str, smtp_port: int):
        self.smtp_host = smtp_host
        self.smtp_port = smtp_port
    
    def send_confirmation(self, email: str) -> bool:
        # Реальная отправка письма
        import smtplib
        try:
            with smtplib.SMTP(self.smtp_host, self.smtp_port) as server:
                server.send_message(f"Confirm your email: {email}")
            return True
        except Exception:
            return False

# Адаптер для HTTP API (In Adapter)
from fastapi import FastAPI, HTTPException

app = FastAPI()

use_case = CreateUserUseCase(
    user_repository=SQLAlchemyUserRepository(db_session),
    email_service=SMTPEmailService("smtp.gmail.com", 587)
)

@app.post("/users")
def create_user_endpoint(username: str, email: str):
    try:
        user = use_case.create_user(username, email)
        return {"id": user.id, "username": user.username, "email": user.email}
    except ValueError as e:
        raise HTTPException(status_code=400, detail=str(e))
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

@app.get("/users/{user_id}")
def get_user_endpoint(user_id: int):
    user = use_case.get_user(user_id)
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    return {"id": user.id, "username": user.username, "email": user.email}

Шаг 4: Тестирование (основное преимущество)

import pytest
from unittest.mock import Mock, MagicMock

# Mock адаптеры для тестирования
class InMemoryUserRepository(UserRepositoryPort):
    def __init__(self):
        self.users: Dict[int, User] = {}
    
    def save(self, user: User) -> None:
        self.users[user.id] = user
    
    def find_by_id(self, user_id: int) -> Optional[User]:
        return self.users.get(user_id)
    
    def find_all(self) -> List[User]:
        return list(self.users.values())

class MockEmailService(EmailServicePort):
    def __init__(self):
        self.sent_emails = []
    
    def send_confirmation(self, email: str) -> bool:
        self.sent_emails.append(email)
        return True

# Тесты
def test_create_user_success():
    # Подготовка
    repository = InMemoryUserRepository()
    email_service = MockEmailService()
    use_case = CreateUserUseCase(repository, email_service)
    
    # Выполнение
    user = use_case.create_user("john_doe", "john@example.com")
    
    # Проверка
    assert user.username == "john_doe"
    assert user.email == "john@example.com"
    assert len(email_service.sent_emails) == 1
    assert email_service.sent_emails[0] == "john@example.com"

def test_create_user_invalid_email():
    repository = InMemoryUserRepository()
    email_service = MockEmailService()
    use_case = CreateUserUseCase(repository, email_service)
    
    with pytest.raises(ValueError):
        use_case.create_user("john_doe", "invalid-email")

Преимущества гексагональной архитектуры

1. Тестируемость — ядро тестируется без реальных зависимостей
2. Независимость фреймворка — можно менять FastAPI на Flask
3. Полнота — все зависимости явные, через конструктор
4. Пластичность — легко добавлять новые адаптеры
5. Чистота кода — бизнес-логика отделена

Недостатки

1. Больше кода — требует больше классов и интерфейсов
2. Начальная сложность — для простых приложений может быть overkill
3. Кривая обучения — нужно понимать паттерны

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