Где находится порт в гексагональной архитектуре?

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

Порты находятся на границе бизнес-логики и внешних систем. Они расположены на уровне Application (Приложение) и выступают промежуточным звеном между ядром системы и адаптерами.

Структура расположения портов

Порты находятся между адаптерами (которые общаются с внешним миром) и ядром системы (доменная логика и случаи использования). Они определяют контракты для взаимодействия.

Два типа портов

1. Входящие порты (Driving Ports) — определяют, как внешние системы могут запросить услугу:

from abc import ABC, abstractmethod

class UserRepositoryPort(ABC):
    """Входящий порт для работы с данными пользователя"""
    
    @abstractmethod
    def find_by_id(self, user_id: int) -> dict:
        pass
    
    @abstractmethod
    def save(self, user_data: dict) -> int:
        pass

2. Исходящие порты (Driven Ports) — определяют, как система обращается во внешние сервисы:

from abc import ABC, abstractmethod

class EmailNotificationPort(ABC):
    """Исходящий порт для отправки email"""
    
    @abstractmethod
    def send_email(self, to: str, subject: str, body: str) -> bool:
        pass

class PaymentProcessorPort(ABC):
    """Исходящий порт для обработки платежей"""
    
    @abstractmethod
    def process_payment(self, amount: float, token: str) -> str:
        pass

Адаптеры имплементируют порты

Адаптеры — это конкретные реализации портов для работы с реальными системами:

from sqlalchemy.orm import Session
from domain.ports.outgoing import UserRepositoryPort

class SQLAlchemyUserRepository(UserRepositoryPort):
    """Адаптер для PostgreSQL"""
    
    def __init__(self, db: Session):
        self.db = db
    
    def find_by_id(self, user_id: int) -> dict:
        user = self.db.query(User).filter(User.id == user_id).first()
        return user.to_dict() if user else None
    
    def save(self, user_data: dict) -> int:
        user = User(**user_data)
        self.db.add(user)
        self.db.commit()
        return user.id

class SendGridEmailAdapter(EmailNotificationPort):
    """Адаптер для SendGrid"""
    
    def __init__(self, api_key: str):
        self.api_key = api_key
    
    def send_email(self, to: str, subject: str, body: str) -> bool:
        # Отправка через SendGrid API
        return True

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

Ядро зависит от портов (интерфейсов), но не от конкретных реализаций. Это позволяет легко менять адаптеры без изменения бизнес-логики:

class RegisterUserUseCase:
    def __init__(
        self, 
        user_repo: UserRepositoryPort,
        email_notifier: EmailNotificationPort
    ):
        self.user_repo = user_repo
        self.email_notifier = email_notifier
    
    def execute(self, username: str, email: str, password: str):
        # Бизнес-логика не знает о реальных реализациях
        user_id = self.user_repo.save({
            'username': username,
            'email': email,
            'password': hash_password(password)
        })
        
        self.email_notifier.send_email(
            to=email,
            subject='Welcome!',
            body=f'Hello, {username}!'
        )
        
        return {'user_id': user_id, 'status': 'registered'}

Практический пример в FastAPI

from fastapi import APIRouter, Depends
from application.use_cases import RegisterUserUseCase

router = APIRouter()

@router.post('/users/register')
async def register_user(
    username: str,
    email: str,
    password: str,
    use_case: RegisterUserUseCase = Depends(get_register_use_case)
):
    result = use_case.execute(username, email, password)
    return result

Ключевые моменты

• Порты на границе ядра и внешнего мира — определяют контракты
• Входящие порты — как система получает запросы
• Исходящие порты — как система отправляет данные
• Адаптеры реализуют порты — переводят абстракцию в конкретику
• Инверсия зависимостей — ядро зависит от интерфейсов, не от реализаций
• Тестируемость — можно легко создавать mock-адаптеры для тестов
• Слабая связанность — замена реализации не требует изменения бизнес-логики