Что такое порт?

Порты в архитектуре

Порт — это интерфейс (абстракция), определяющий контракт между ядром приложения и внешним миром. Это ключевой концепт Hexagonal Architecture для обеспечения независимости от фреймворков и технологий.

Входящие порты (Inbound Ports)

Это интерфейсы use cases, через которые внешний мир запускает действия:

from abc import ABC, abstractmethod
import uuid

class CreateUserPort(ABC):
    @abstractmethod
    def execute(self, email: str, name: str) -> str:
        pass

class CreateUserUseCase(CreateUserPort):
    def __init__(self, repository: "UserRepository"):
        self.repository = repository

    def execute(self, email: str, name: str) -> str:
        user_id = str(uuid.uuid4())
        self.repository.save(user_id, email, name)
        return user_id

Выходящие порты (Outbound Ports)

Это интерфейсы для работы с внешними сервисами (БД, API). Ядро зависит от абстракции:

class UserRepository(ABC):
    @abstractmethod
    def save(self, user_id: str, email: str, name: str) -> None:
        pass

    @abstractmethod
    def get_by_email(self, email: str) -> dict | None:
        pass

class PostgresUserRepository(UserRepository):
    def __init__(self, db):
        self.db = db

    def save(self, user_id: str, email: str, name: str) -> None:
        self.db.execute(
            "INSERT INTO users (id, email, name) VALUES (%s, %s, %s)",
            (user_id, email, name)
        )

    def get_by_email(self, email: str) -> dict | None:
        return self.db.fetch_one(
            "SELECT * FROM users WHERE email = %s", (email,)
        )

class MongoUserRepository(UserRepository):
    def __init__(self, mongo_client):
        self.db = mongo_client.myapp

    def save(self, user_id: str, email: str, name: str) -> None:
        self.db.users.insert_one({"id": user_id, "email": email, "name": name})

    def get_by_email(self, email: str) -> dict | None:
        return self.db.users.find_one({"email": email})

Адаптеры REST API

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

from fastapi import APIRouter

router = APIRouter()
create_user_use_case: CreateUserUseCase

@router.post("/users")
async def create_user(email: str, name: str):
    user_id = create_user_use_case.execute(email, name)
    return {"id": user_id}

Тестирование с Mock адаптерами

class FakeRepository(UserRepository):
    def __init__(self):
        self.users = {}

    def save(self, user_id: str, email: str, name: str) -> None:
        self.users[user_id] = {"id": user_id, "email": email, "name": name}

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

def test_create_user():
    repo = FakeRepository()
    use_case = CreateUserUseCase(repo)
    user_id = use_case.execute("john@example.com", "John")
    assert user_id in repo.users
    assert repo.users[user_id]["email"] == "john@example.com"

Преимущества портов и адаптеров

• Независимость от фреймворков — легко заменить FastAPI на Flask
• Независимость от БД — переход PostgreSQL → MongoDB без изменения use case
• Тестируемость — используем mock адаптеры в unit-тестах
• Чистая архитектура — границы между слоями чёткие
• SOLID-принципы — особенно Dependency Inversion

Основная идея: ядро (domain + use cases) не знает о внешних технологиях. Все внешние зависимости выражены через портовые интерфейсы и внедрены как адаптеры.