Hexagonal Architecture란 무엇일까

[UI Layer] → [Business Logic Layer] → [Data Access Layer]

        애플리케이션을 쓰는 쪽         애플리케이션이 쓰는 쪽
        (Driver / Primary)           (Driven / Secondary)
        ─────────────────            ─────────────────
        HTTP 요청                     PostgreSQL
        CLI 명령어                    Redis 캐시
        테스트 코드                   이메일 서비스
        메시지 큐                     외부 API
                    ↘               ↙
                  [ Application Core ]

                    ┌─────────────────────┐
                    │                     │
     Driver ──Port──│   Application Core  │──Port──▶ Driven
    (Primary)       │                     │        (Secondary)
                    │  ┌───────────────┐  │
                    │  │ Domain Model  │  │
                    │  │ App Services  │  │
                    │  └───────────────┘  │
                    │                     │
                    └─────────────────────┘

Port: OrderRepository
├── Adapter: PostgresOrderRepository    (운영용)
├── Adapter: InMemoryOrderRepository    (테스트용)
└── Adapter: MongoOrderRepository       (마이그레이션 중)

Port: PlacingOrders
├── Adapter: RestApiController          (HTTP)
├── Adapter: GraphQLResolver            (GraphQL)
├── Adapter: CLIHandler                 (커맨드라인)
└── Adapter: TestHarness                (자동화 테스트)

[REST Controller] ──uses──▶ [PlacingOrders Port] ──▶ [Application Core]
[Test Harness]    ──uses──▶ [PlacingOrders Port] ──▶ [Application Core]
[CLI Command]     ──uses──▶ [PlacingOrders Port] ──▶ [Application Core]

[Application Core] ──uses──▶ [OrderRepository Port]
                                      │
                              implements│
                                      ▼
                          [PostgresOrderRepository Adapter]

order_service/
├── domain/                    # 도메인 모델 (순수 비즈니스 로직)
│   ├── model.py               # Entity, Value Object
│   └── ports.py               # 포트 (인터페이스) 정의
├── application/               # 유스케이스 오케스트레이션
│   └── service.py             # Application Service
├── adapters/                  # 어댑터 (기술 구현)
│   ├── inbound/               # Driver(Primary) 어댑터
│   │   └── api.py             # FastAPI REST 컨트롤러
│   └── outbound/              # Driven(Secondary) 어댑터
│       ├── postgres_repo.py   # PostgreSQL 구현
│       └── memory_repo.py     # In-Memory 구현 (테스트용)
└── config.py                  # 의존성 조립 (Composition Root)

# domain/model.py
from dataclasses import dataclass
from enum import Enum
from uuid import UUID, uuid4


class OrderStatus(Enum):
    PENDING = "pending"
    CONFIRMED = "confirmed"
    CANCELLED = "cancelled"


@dataclass
class Order:
    id: UUID
    product: str
    quantity: int
    status: OrderStatus

    @staticmethod
    def create(product: str, quantity: int) -> "Order":
        if quantity <= 0:
            raise ValueError("수량은 1 이상이어야 합니다")
        return Order(
            id=uuid4(),
            product=product,
            quantity=quantity,
            status=OrderStatus.PENDING,
        )

    def confirm(self) -> None:
        if self.status != OrderStatus.PENDING:
            raise ValueError(f"{self.status.value} 상태의 주문은 확정할 수 없습니다")
        self.status = OrderStatus.CONFIRMED

    def cancel(self) -> None:
        if self.status == OrderStatus.CONFIRMED:
            raise ValueError("이미 확정된 주문은 취소할 수 없습니다")
        self.status = OrderStatus.CANCELLED

# domain/ports.py
from abc import ABC, abstractmethod
from uuid import UUID
from domain.model import Order


# ── Driven(Secondary) Port ──────────────────────────────
# 애플리케이션이 "주문을 저장/조회하겠다"는 conversation을 정의
class OrderRepository(ABC):
    @abstractmethod
    def save(self, order: Order) -> None: ...

    @abstractmethod
    def find_by_id(self, order_id: UUID) -> Order | None: ...


# ── Driven(Secondary) Port ──────────────────────────────
# 애플리케이션이 "알림을 보내겠다"는 conversation을 정의
class NotificationSender(ABC):
    @abstractmethod
    def send(self, message: str) -> None: ...


# ── Driver(Primary) Port ────────────────────────────────
# 외부가 "주문을 다루겠다"는 conversation을 정의
class OrderUseCase(ABC):
    @abstractmethod
    def place_order(self, product: str, quantity: int) -> Order: ...

    @abstractmethod
    def confirm_order(self, order_id: UUID) -> Order: ...

# application/service.py
from uuid import UUID
from domain.model import Order
from domain.ports import OrderRepository, NotificationSender, OrderUseCase


class OrderService(OrderUseCase):
    def __init__(
        self,
        repo: OrderRepository,           # Driven 포트에 의존
        notifier: NotificationSender,     # Driven 포트에 의존
    ):
        self._repo = repo
        self._notifier = notifier

    def place_order(self, product: str, quantity: int) -> Order:
        order = Order.create(product, quantity)   # 도메인 로직
        self._repo.save(order)                    # 포트를 통해 저장
        return order

    def confirm_order(self, order_id: UUID) -> Order:
        order = self._repo.find_by_id(order_id)
        if order is None:
            raise ValueError("주문을 찾을 수 없습니다")
        order.confirm()                           # 도메인 로직
        self._repo.save(order)                    # 포트를 통해 저장
        self._notifier.send(                      # 포트를 통해 알림
            f"주문 {order.id} 확정됨"
        )
        return order

# adapters/outbound/memory_repo.py
from uuid import UUID
from domain.model import Order
from domain.ports import OrderRepository


class InMemoryOrderRepository(OrderRepository):
    def __init__(self):
        self._store: dict[UUID, Order] = {}

    def save(self, order: Order) -> None:
        self._store[order.id] = order

    def find_by_id(self, order_id: UUID) -> Order | None:
        return self._store.get(order_id)

# adapters/outbound/postgres_repo.py
from uuid import UUID
from domain.model import Order
from domain.ports import OrderRepository


class PostgresOrderRepository(OrderRepository):
    def __init__(self, connection_string: str):
        self._conn_str = connection_string
        # 실제 DB 커넥션 설정...

    def save(self, order: Order) -> None:
        # INSERT/UPDATE SQL 실행
        ...

    def find_by_id(self, order_id: UUID) -> Order | None:
        # SELECT SQL 실행
        ...

# adapters/inbound/api.py  (Driver Adapter)
from uuid import UUID
from fastapi import FastAPI, Depends, HTTPException
from domain.ports import OrderUseCase
from config import create_order_service

app = FastAPI()


def get_order_use_case() -> OrderUseCase:
    return create_order_service()


@app.post("/orders")
def create_order(
    product: str,
    quantity: int,
    use_case: OrderUseCase = Depends(get_order_use_case),
):
    try:
        order = use_case.place_order(product, quantity)
        return {"id": str(order.id), "status": order.status.value}
    except ValueError as e:
        raise HTTPException(status_code=400, detail=str(e))


@app.post("/orders/{order_id}/confirm")
def confirm_order(
    order_id: UUID,
    use_case: OrderUseCase = Depends(get_order_use_case),
):
    try:
        order = use_case.confirm_order(order_id)
        return {"id": str(order.id), "status": order.status.value}
    except ValueError as e:
        raise HTTPException(status_code=400, detail=str(e))

# config.py
from application.service import OrderService
from adapters.outbound.memory_repo import InMemoryOrderRepository
from adapters.outbound.postgres_repo import PostgresOrderRepository
import os


def create_order_service() -> OrderService:
    env = os.getenv("ENV", "dev")

    if env == "test":
        repo = InMemoryOrderRepository()
    else:
        repo = PostgresOrderRepository(os.getenv("DATABASE_URL"))

    notifier = ConsoleNotifier()  # 또는 EmailNotifier, SlackNotifier...

    return OrderService(repo=repo, notifier=notifier)

# tests/test_order_service.py
from application.service import OrderService
from adapters.outbound.memory_repo import InMemoryOrderRepository


class FakeNotifier(NotificationSender):
    def __init__(self):
        self.messages = []

    def send(self, message: str) -> None:
        self.messages.append(message)


def test_place_and_confirm_order():
    repo = InMemoryOrderRepository()
    notifier = FakeNotifier()
    service = OrderService(repo=repo, notifier=notifier)

    order = service.place_order("커피", 2)
    assert order.status.value == "pending"

    confirmed = service.confirm_order(order.id)
    assert confirmed.status.value == "confirmed"
    assert len(notifier.messages) == 1

          ┌────────────────────────────────────────────────────────┐
          │                                                        │
 Driver   │   ┌──────────────────────────────────────────┐         │  Driven
 Adapters │   │            Application Core              │         │  Adapters
          │   │                                          │         │
 REST ────┼──▶│  OrderUseCase ◁── OrderService           │         │
 CLI  ────┼──▶│                    │    │                 │         │
 Test ────┼──▶│                    │    │                 │         │
          │   │                    ▼    ▼                 │         │
          │   │         OrderRepository  NotificationSender│        │
          │   │              ▲                ▲           │         │
          │   └──────────────┼────────────────┼──────────┘         │
          │                  │                │                    │
          │           Postgres Repo     Email Notifier            │
          │           Memory Repo       Slack Notifier            │
          └────────────────────────────────────────────────────────┘
          모든 화살표가 안쪽(Core)을 향합니다.

# 추천 구조 (기능/모듈 단위)
order/
├── domain/
│   ├── model.py
│   └── ports.py
├── application/
│   └── service.py
└── adapters/
    ├── inbound/
    └── outbound/

# 또는 (계층 단위 — 소규모 프로젝트)
domain/
application/
adapters/
    inbound/
    outbound/

# BAD — 너무 큰 포트 하나에 다 몰아넣음
class DatabasePort(ABC):
    def save_order(self, order): ...
    def find_order(self, id): ...
    def save_user(self, user): ...
    def find_user(self, id): ...

# GOOD — conversation(목적) 단위로 분리
class OrderRepository(ABC):
    def save(self, order): ...
    def find_by_id(self, id): ...

class UserRepository(ABC):
    def save(self, user): ...
    def find_by_id(self, id): ...

# BAD — 비즈니스 로직이 서비스에 있음
class OrderService:
    def confirm_order(self, order_id):
        order = self.repo.find_by_id(order_id)
        if order.status != "pending":                    # 이 로직은
            raise ValueError("확정 불가")                  # 도메인에 있어야 한다
        order.status = "confirmed"
        self.repo.save(order)

# GOOD — 도메인 모델에 비즈니스 규칙이 캡슐화됨
class OrderService:
    def confirm_order(self, order_id):
        order = self.repo.find_by_id(order_id)
        order.confirm()                                  # 도메인이 판단
        self.repo.save(order)

# BAD — 상태를 물어서 밖에서 판단 (Ask)
if order.status == OrderStatus.PENDING:
    order.status = OrderStatus.CONFIRMED

# GOOD — 행위를 지시 (Tell)
order.confirm()  # 판단 로직이 도메인 안에 캡슐화됨

# BAD — Application Service에 HTTP 기술이 침투함
class OrderService:
    def place_order(self, request_json: dict):  # HTTP 세부사항이 여기까지 옴
        product = request_json["product"]       # JSON 파싱이 서비스 안에 있음
        ...

# GOOD — Application Service는 기술 무관
class OrderService:
    def place_order(self, product: str, quantity: int):  # 순수한 파라미터만 받음
        ...
# JSON 파싱은 Driver Adapter(REST Controller)가 처리합니다

# BAD — Use Case가 다른 Use Case를 호출
class ConfirmOrderService:
    def __init__(self, notify_service: NotifyService):  # 다른 서비스에 의존
        self._notify = notify_service

    def confirm(self, order_id):
        ...
        self._notify.send_confirmation(order_id)  # 서비스 간 커플링

비즈니스 규칙이 복잡한가?
├── No → Layered Architecture로 충분
└── Yes ↓
    외부 시스템 교체 가능성이 높은가?
    ├── Yes → Hexagonal Architecture ✓
    └── No ↓
        도메인 모델이 핵심 복잡성인가?
        ├── Yes → Onion Architecture (DDD 결합)
        └── No → Clean Architecture (범용)