Как понять, что код относится к гексагональной архитектуре?

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

Гексагональная архитектура (также называется ports and adapters архитектурой) — это паттерн проектирования, который изолирует ядро бизнес-логики от деталей реализации (БД, API, framework). Вот признаки правильной гексагональной архитектуры:

1. Структура проекта

project/
├── domain/                      # Ядро, бизнес-логика
│   ├── models/
│   │   ├── user.py             # Entity (доменная модель)
│   │   ├── order.py
│   │   └── value_objects.py    # Value Objects
│   └── exceptions.py           # Доменные исключения
│
├── application/                # Логика приложения (use cases)
│   ├── services/
│   │   ├── user_service.py    # Бизнес случаи использования
│   │   └── order_service.py
│   ├── dto/                   # Data Transfer Objects
│   │   ├── user_dto.py
│   │   └── order_dto.py
│   └── ports/                 # Интерфейсы (contracts)
│       ├── user_repository.py  # Port (интерфейс)
│       ├── payment_gateway.py
│       └── notification_port.py
│
├── infrastructure/            # Реализация деталей
│   ├── repositories/          # Adapters для БД
│   │   ├── postgres_user_repo.py
│   │   └── postgres_order_repo.py
│   ├── gateways/             # Adapters для внешних сервисов
│   │   ├── stripe_payment_gateway.py
│   │   └── email_notification.py
│   ├── database/             # БД конфигурация
│   │   ├── models.py         # ORM модели
│   │   └── connection.py
│   └── config/               # Конфигурация
│       └── settings.py
│
└── presentation/             # Пользовательский интерфейс
    ├── api/                  # HTTP endpoints
    │   ├── routes/
    │   │   ├── users.py
    │   │   └── orders.py
    │   └── dependencies.py   # FastAPI dependencies
    ├── cli/                  # Командная строка
    │   └── commands.py
    └── web/                  # Веб интерфейс (опционально)
        └── templates/

2. Слои архитектуры (зависимости идут только внутрь)

        ВНЕШНИЙ МИР
        ├─ HTTP (API)
        ├─ БД (PostgreSQL)
        ├─ внешние API (Stripe, Email)
        └─ CLI
            ↓
    ┌─────────────────────────┐
    │   PRESENTATION LAYER    │ ← Входные точки (API routes, CLI)
    │  (HTTP, CLI, Web UI)    │
    └──────────┬──────────────┘
               ↓ (зависит от)
    ┌─────────────────────────┐
    │  APPLICATION LAYER      │ ← Бизнес случаи (use cases)
    │  (Services, Use Cases)  │   Зависит от ports
    └──────────┬──────────────┘
               ↓ (зависит от)
    ┌─────────────────────────┐
    │  DOMAIN LAYER (Ядро)    │ ← Бизнес-логика
    │ (Entities, Value Obj)   │   НЕ зависит ни от чего!
    └─────────────────────────┘
               ↑ (реализуют)
    ┌─────────────────────────┐
    │  INFRASTRUCTURE LAYER   │ ← Реализация деталей
    │ (DB, APIs, Adapters)   │   (repositories, gateways)
    └─────────────────────────┘

3. Domain Layer (Ядро)

Доменная сущность (не зависит от БД)

# domain/models/user.py
from dataclasses import dataclass
from datetime import datetime
from enum import Enum

class UserStatus(str, Enum):
    ACTIVE = "active"
    INACTIVE = "inactive"
    BANNED = "banned"

@dataclass
class User:
    """Доменная сущность User."""
    id: int
    email: str
    name: str
    status: UserStatus
    created_at: datetime
    
    def deactivate(self):
        """Бизнес-логика: деактивация пользователя."""
        if self.status == UserStatus.BANNED:
            raise DomainError("Cannot deactivate banned user")
        self.status = UserStatus.INACTIVE
    
    def is_active(self) -> bool:
        return self.status == UserStatus.ACTIVE

# domain/exceptions.py
class DomainError(Exception):
    """Базовое доменное исключение."""
    pass

class UserNotFoundError(DomainError):
    pass

class InvalidEmailError(DomainError):
    pass

Ключевой момент: User — это просто данные + бизнес-логика, БЕЗ SQL, БЕЗ HTTP, БЕЗ фреймворков.

4. Ports (Интерфейсы)

Port — это контракт (interface) для внешних зависимостей

# application/ports/user_repository.py
from abc import ABC, abstractmethod
from domain.models.user import User
from typing import Optional

class UserRepository(ABC):
    """Port: интерфейс для хранилища пользователей."""
    
    @abstractmethod
    async def save(self, user: User) -> None:
        """Сохранить пользователя."""
        pass
    
    @abstractmethod
    async def find_by_id(self, user_id: int) -> Optional[User]:
        """Найти пользователя по ID."""
        pass
    
    @abstractmethod
    async def find_by_email(self, email: str) -> Optional[User]:
        """Найти пользователя по email."""
        pass
    
    @abstractmethod
    async def delete(self, user_id: int) -> None:
        """Удалить пользователя."""
        pass

# application/ports/payment_gateway.py
class PaymentGateway(ABC):
    """Port: интерфейс для обработки платежей."""
    
    @abstractmethod
    async def charge(self, amount: float, card_token: str) -> str:
        """Charge -> transaction_id"""
        pass
    
    @abstractmethod
    async def refund(self, transaction_id: str) -> bool:
        pass

# application/ports/notification_port.py
class NotificationPort(ABC):
    """Port: интерфейс для отправки уведомлений."""
    
    @abstractmethod
    async def send_email(self, to: str, subject: str, body: str) -> None:
        pass

Замечание: Ports — это абстракции, они определяют, что нужно, но не как это делать.

5. Application Layer (Use Cases)

# application/services/user_service.py
from domain.models.user import User, UserStatus, InvalidEmailError
from domain.exceptions import UserNotFoundError
from application.ports.user_repository import UserRepository
from application.ports.notification_port import NotificationPort
from application.dto.user_dto import CreateUserDTO, UserResponseDTO

class UserService:
    """Use case: управление пользователями."""
    
    def __init__(
        self,
        user_repository: UserRepository,  # Injection: зависимость
        notification: NotificationPort
    ):
        self.user_repository = user_repository
        self.notification = notification
    
    async def create_user(self, dto: CreateUserDTO) -> UserResponseDTO:
        """Use case: создание пользователя."""
        # Валидация
        if "@" not in dto.email:
            raise InvalidEmailError("Invalid email")
        
        # Проверяем, не существует ли уже
        existing = await self.user_repository.find_by_email(dto.email)
        if existing:
            raise DomainError("User already exists")
        
        # Создаём доменную сущность
        user = User(
            id=None,  # БД сгенерирует
            email=dto.email,
            name=dto.name,
            status=UserStatus.ACTIVE,
            created_at=datetime.utcnow()
        )
        
        # Сохраняем (через port)
        await self.user_repository.save(user)
        
        # Отправляем уведомление (через port)
        await self.notification.send_email(
            to=user.email,
            subject="Welcome!",
            body=f"Hello {user.name}"
        )
        
        return UserResponseDTO.from_domain(user)
    
    async def get_user(self, user_id: int) -> UserResponseDTO:
        """Use case: получение пользователя."""
        user = await self.user_repository.find_by_id(user_id)
        if not user:
            raise UserNotFoundError(f"User {user_id} not found")
        return UserResponseDTO.from_domain(user)
    
    async def deactivate_user(self, user_id: int) -> UserResponseDTO:
        """Use case: деактивация пользователя."""
        user = await self.user_repository.find_by_id(user_id)
        if not user:
            raise UserNotFoundError(f"User {user_id} not found")
        
        # Вызываем бизнес-логику доменной сущности
        user.deactivate()  # Может выбросить исключение
        
        # Сохраняем изменения
        await self.user_repository.save(user)
        
        return UserResponseDTO.from_domain(user)

# application/dto/user_dto.py
from dataclasses import dataclass
from domain.models.user import User

@dataclass
class CreateUserDTO:
    """DTO для входящего запроса."""
    email: str
    name: str

@dataclass
class UserResponseDTO:
    """DTO для исходящего ответа."""
    id: int
    email: str
    name: str
    status: str
    created_at: str
    
    @classmethod
    def from_domain(cls, user: User) -> "UserResponseDTO":
        """Преобразование доменной сущности в DTO."""
        return cls(
            id=user.id,
            email=user.email,
            name=user.name,
            status=user.status.value,
            created_at=user.created_at.isoformat()
        )

Ключевой момент: Service — это оркестратор, он:

1. Получает данные через ports
2. Вызывает доменную логику
3. Сохраняет результат через ports

6. Infrastructure Layer (Adapters)

Adapter для БД

# infrastructure/repositories/postgres_user_repo.py
from application.ports.user_repository import UserRepository
from domain.models.user import User, UserStatus
from infrastructure.database.models import UserORM
from sqlalchemy.ext.asyncio import AsyncSession

class PostgresUserRepository(UserRepository):
    """Adapter: реализация repository для PostgreSQL."""
    
    def __init__(self, db_session: AsyncSession):
        self.db = db_session
    
    async def save(self, user: User) -> None:
        """Реализация сохранения в PostgreSQL."""
        user_orm = UserORM(
            id=user.id,
            email=user.email,
            name=user.name,
            status=user.status.value,
            created_at=user.created_at
        )
        self.db.add(user_orm)
        await self.db.flush()
    
    async def find_by_id(self, user_id: int) -> Optional[User]:
        """Реализация поиска в PostgreSQL."""
        result = await self.db.execute(
            select(UserORM).where(UserORM.id == user_id)
        )
        user_orm = result.scalar()
        return self._orm_to_domain(user_orm) if user_orm else None
    
    def _orm_to_domain(self, user_orm: UserORM) -> User:
        """Преобразование ORM модели в доменную сущность."""
        return User(
            id=user_orm.id,
            email=user_orm.email,
            name=user_orm.name,
            status=UserStatus(user_orm.status),
            created_at=user_orm.created_at
        )

# infrastructure/database/models.py
from sqlalchemy import Column, Integer, String, DateTime
from datetime import datetime

class UserORM(Base):
    """ORM модель (ТОЛЬКО для БД, не в domain!)."""
    __tablename__ = "users"
    
    id = Column(Integer, primary_key=True)
    email = Column(String, unique=True, nullable=False)
    name = Column(String, nullable=False)
    status = Column(String, nullable=False)
    created_at = Column(DateTime, default=datetime.utcnow)

Adapter для внешнего сервиса

# infrastructure/gateways/stripe_payment_gateway.py
from application.ports.payment_gateway import PaymentGateway
import stripe

class StripePaymentGateway(PaymentGateway):
    """Adapter: Stripe для обработки платежей."""
    
    def __init__(self, api_key: str):
        stripe.api_key = api_key
    
    async def charge(self, amount: float, card_token: str) -> str:
        """Реализация платежа через Stripe."""
        charge = stripe.Charge.create(
            amount=int(amount * 100),
            currency="usd",
            source=card_token
        )
        return charge.id
    
    async def refund(self, transaction_id: str) -> bool:
        stripe.Refund.create(charge=transaction_id)
        return True

# infrastructure/gateways/email_notification.py
from application.ports.notification_port import NotificationPort
import aiosmtplib

class EmailNotificationGateway(NotificationPort):
    """Adapter: Email для отправки уведомлений."""
    
    def __init__(self, smtp_host: str, smtp_port: int):
        self.smtp_host = smtp_host
        self.smtp_port = smtp_port
    
    async def send_email(self, to: str, subject: str, body: str) -> None:
        """Реализация отправки email."""
        async with aiosmtplib.SMTP(hostname=self.smtp_host, port=self.smtp_port) as smtp:
            await smtp.send_message(
                f"To: {to}\nSubject: {subject}\n\n{body}"
            )

7. Presentation Layer (API)

# presentation/api/routes/users.py
from fastapi import APIRouter, Depends, HTTPException, status
from application.services.user_service import UserService
from application.dto.user_dto import CreateUserDTO, UserResponseDTO
from domain.exceptions import DomainError, UserNotFoundError
from presentation.api.dependencies import get_user_service

router = APIRouter(prefix="/users", tags=["users"])

@router.post("", response_model=UserResponseDTO, status_code=status.HTTP_201_CREATED)
async def create_user(
    dto: CreateUserDTO,
    service: UserService = Depends(get_user_service)
):
    """Endpoint: создание пользователя."""
    try:
        return await service.create_user(dto)
    except DomainError as e:
        raise HTTPException(status_code=400, detail=str(e))

@router.get("/{user_id}", response_model=UserResponseDTO)
async def get_user(
    user_id: int,
    service: UserService = Depends(get_user_service)
):
    try:
        return await service.get_user(user_id)
    except UserNotFoundError:
        raise HTTPException(status_code=404, detail="User not found")

# presentation/api/dependencies.py
from fastapi import Depends
from sqlalchemy.ext.asyncio import AsyncSession
from application.services.user_service import UserService
from infrastructure.repositories.postgres_user_repo import PostgresUserRepository
from infrastructure.gateways.email_notification import EmailNotificationGateway
from infrastructure.database.connection import get_db_session

async def get_user_service(db: AsyncSession = Depends(get_db_session)) -> UserService:
    """Dependency Injection: собираем сервис с зависимостями."""
    repository = PostgresUserRepository(db)
    notification = EmailNotificationGateway(
        smtp_host="smtp.gmail.com",
        smtp_port=587
    )
    return UserService(repository, notification)

8. Признаки правильной гексагональной архитектуры

✅ Правильно:

• Domain не знает о FastAPI, SQLAlchemy, Stripe
• Зависимости идут только от внешних слоёв к внутренним
• Каждый layer имеет одну ответственность
• Ports — абстракции (interfaces)
• Adapters — конкретные реализации
• DTOs для преобразования между слоями
• Тестирование domain без БД

❌ Неправильно:

• Domain импортирует SQLAlchemy, FastAPI
• Циклические зависимости между layers
• Domain зависит от Adapters
• Прямой доступ к БД из API
• Смешивание logic разных layers

9. Тестирование (преимущество гексагональной архитектуры)

# tests/unit/test_user_service.py
import pytest
from unittest.mock import AsyncMock
from application.services.user_service import UserService
from application.dto.user_dto import CreateUserDTO
from domain.models.user import User, UserStatus
from domain.exceptions import DomainError

class MockUserRepository:
    async def save(self, user: User):
        pass
    
    async def find_by_email(self, email: str):
        return None

class MockNotification:
    async def send_email(self, to: str, subject: str, body: str):
        pass

@pytest.mark.asyncio
async def test_create_user():
    # Arrange
    repo = MockUserRepository()
    notification = MockNotification()
    service = UserService(repo, notification)
    dto = CreateUserDTO(email="john@example.com", name="John")
    
    # Act
    result = await service.create_user(dto)
    
    # Assert
    assert result.email == "john@example.com"
    assert result.status == "active"

Ключевой момент: Можно тестировать Service БЕЗ БД и БЕЗ HTTP, используя mock'и!

Итоговая таблица

Гексагональная архитектура = Testability + Flexibility + Clean Code!