Hexagonal Architecture(포트 & 어댑터) 실전 가이드 — 클린 아키텍처의 핵심

1. 도메인 (Domain / Core)
   - 비즈니스 로직의 핵심
   - 외부 의존성 없음 (순수 코드)
   - Entity, Value Object, Domain Service

2. 포트 (Port)
   - 도메인과 외부 세계의 인터페이스(계약)
   - Input Port: 외부 → 도메인 (Use Case)
   - Output Port: 도메인 → 외부 (Repository 인터페이스)

3. 어댑터 (Adapter)
   - 포트의 구체적 구현
   - Input Adapter: REST Controller, gRPC Handler, CLI
   - Output Adapter: MySQL Repository, Redis Cache, HTTP Client

# Layered Architecture (전통적)
# Controller → Service → Repository → DB
# 의존성 방향: 위 → 아래 (DB에 종속)

# Hexagonal Architecture
# Adapter → Port → Domain ← Port ← Adapter
# 의존성 방향: 바깥 → 안쪽 (Domain이 중심)

[임포트 방향]   adapters ──import──▶ ports ──import──▶ domain
                (바깥)                                  (안쪽)

[호출 방향]     domain ──call──▶ ports(추상 타입) ──▶ adapters
                (안쪽)                                  (바깥)

같은 경계를 두 화살표가 서로 반대로 지나간다.
이 반대 방향이 곧 의존성 역전이다.

# domain/models/order.py  ← 도메인인데 인프라를 import 한다
from sqlalchemy.orm import Session           # ORM 세션 타입
from infrastructure.db import OrderTable     # ORM 매핑 클래스


class Order:
    def confirm(self, session: Session) -> None:
        if not self.items:
            raise ValueError("상품이 없는 주문은 확정할 수 없습니다")
        self.status = "confirmed"
        session.merge(OrderTable.from_domain(self))   # 도메인이 커밋을 안다
        session.commit()

주도(primary / driving) 포트 — 바깥이 호출을 시작한다
    REST 컨트롤러 / gRPC 핸들러 / CLI / 배치 잡 / 큐 컨슈머 / 인수 테스트
              │
              ▼
      ┌────────────────────────────┐
      │        애플리케이션          │
      │    (유스케이스 + 도메인)      │
      └────────────────────────────┘
              │
              ▼
피주도(secondary / driven) 포트 — 애플리케이션이 호출을 시작한다
    OrderRepository / PaymentGateway / NotificationSender / Clock

판별법: 이 대화를 누가 먼저 걸었는가?

order-service/
├── domain/                    # 핵심 도메인
│   ├── models/
│   │   ├── order.py          # Entity
│   │   └── order_item.py     # Value Object
│   └── services/
│       └── order_service.py  # Domain Service
├── ports/                     # 포트 (인터페이스)
│   ├── input/
│   │   └── order_use_case.py # Input Port
│   └── output/
│       ├── order_repository.py    # Output Port
│       └── payment_gateway.py     # Output Port
├── adapters/                  # 어댑터 (구현)
│   ├── input/
│   │   ├── rest_controller.py     # REST API
│   │   └── grpc_handler.py        # gRPC
│   └── output/
│       ├── postgres_order_repo.py # PostgreSQL 구현
│       ├── redis_order_cache.py   # Redis 캐시
│       └── stripe_payment.py      # Stripe 결제
└── config/
    └── dependency_injection.py    # DI 설정

# domain/models/order.py
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from typing import List
from uuid import UUID, uuid4


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


@dataclass
class OrderItem:
    product_id: str
    product_name: str
    quantity: int
    unit_price: float

    @property
    def subtotal(self) -> float:
        return self.quantity * self.unit_price


@dataclass
class Order:
    """주문 엔티티 — 비즈니스 규칙을 포함"""
    id: UUID = field(default_factory=uuid4)
    customer_id: str = ""
    items: List[OrderItem] = field(default_factory=list)
    status: OrderStatus = OrderStatus.PENDING
    created_at: datetime = field(default_factory=datetime.now)

    @property
    def total_amount(self) -> float:
        return sum(item.subtotal for item in self.items)

    def add_item(self, item: OrderItem) -> None:
        if self.status != OrderStatus.PENDING:
            raise ValueError("확정된 주문에는 상품을 추가할 수 없습니다")
        if item.quantity <= 0:
            raise ValueError("수량은 1 이상이어야 합니다")
        self.items.append(item)

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

    def cancel(self) -> None:
        if self.status == OrderStatus.SHIPPED:
            raise ValueError("배송된 주문은 취소할 수 없습니다")
        self.status = OrderStatus.CANCELLED

# ports/input/order_use_case.py
from abc import ABC, abstractmethod
from uuid import UUID
from domain.models.order import Order, OrderItem


class CreateOrderUseCase(ABC):
    @abstractmethod
    def execute(self, customer_id: str, items: list[OrderItem]) -> Order:
        pass


class ConfirmOrderUseCase(ABC):
    @abstractmethod
    def execute(self, order_id: UUID) -> Order:
        pass


class CancelOrderUseCase(ABC):
    @abstractmethod
    def execute(self, order_id: UUID) -> Order:
        pass


# ports/output/order_repository.py
from abc import ABC, abstractmethod
from uuid import UUID
from domain.models.order import Order


class OrderRepository(ABC):
    @abstractmethod
    def save(self, order: Order) -> None:
        pass

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

    @abstractmethod
    def find_by_customer(self, customer_id: str) -> list[Order]:
        pass


# ports/output/payment_gateway.py
from abc import ABC, abstractmethod
from uuid import UUID


class PaymentGateway(ABC):
    @abstractmethod
    def charge(self, order_id: UUID, amount: float, customer_id: str) -> bool:
        pass

    @abstractmethod
    def refund(self, order_id: UUID) -> bool:
        pass

# domain/services/order_service.py
from uuid import UUID
from domain.models.order import Order, OrderItem
from ports.input.order_use_case import (
    CreateOrderUseCase, ConfirmOrderUseCase, CancelOrderUseCase
)
from ports.output.order_repository import OrderRepository
from ports.output.payment_gateway import PaymentGateway


class OrderService(CreateOrderUseCase, ConfirmOrderUseCase, CancelOrderUseCase):
    """주문 서비스 — Input Port 구현"""

    def __init__(
        self,
        order_repo: OrderRepository,
        payment_gateway: PaymentGateway
    ):
        # Output Port에 의존 (구체 구현이 아님!)
        self._order_repo = order_repo
        self._payment_gateway = payment_gateway

    def execute(self, customer_id: str = None, items: list[OrderItem] = None,
                order_id: UUID = None) -> Order:
        # dispatch based on params (simplified)
        if customer_id and items:
            return self._create_order(customer_id, items)
        raise ValueError("Invalid parameters")

    def _create_order(self, customer_id: str, items: list[OrderItem]) -> Order:
        order = Order(customer_id=customer_id)
        for item in items:
            order.add_item(item)
        self._order_repo.save(order)
        return order

    def confirm_order(self, order_id: UUID) -> Order:
        order = self._order_repo.find_by_id(order_id)
        if not order:
            raise ValueError(f"주문을 찾을 수 없습니다: {order_id}")

        # 결제 처리
        success = self._payment_gateway.charge(
            order_id=order.id,
            amount=order.total_amount,
            customer_id=order.customer_id
        )
        if not success:
            raise ValueError("결제에 실패했습니다")

        order.confirm()
        self._order_repo.save(order)
        return order

    def cancel_order(self, order_id: UUID) -> Order:
        order = self._order_repo.find_by_id(order_id)
        if not order:
            raise ValueError(f"주문을 찾을 수 없습니다: {order_id}")

        order.cancel()
        self._payment_gateway.refund(order_id)
        self._order_repo.save(order)
        return order

# adapters/output/postgres_order_repo.py
import psycopg2
from uuid import UUID
from domain.models.order import Order, OrderItem, OrderStatus
from ports.output.order_repository import OrderRepository


class PostgresOrderRepository(OrderRepository):
    def __init__(self, connection_string: str):
        self._conn_str = connection_string

    def save(self, order: Order) -> None:
        with psycopg2.connect(self._conn_str) as conn:
            with conn.cursor() as cur:
                cur.execute("""
                    INSERT INTO orders (id, customer_id, status, created_at)
                    VALUES (%s, %s, %s, %s)
                    ON CONFLICT (id) DO UPDATE SET status = %s
                """, (str(order.id), order.customer_id,
                      order.status.value, order.created_at,
                      order.status.value))

                for item in order.items:
                    cur.execute("""
                        INSERT INTO order_items
                        (order_id, product_id, product_name, quantity, unit_price)
                        VALUES (%s, %s, %s, %s, %s)
                        ON CONFLICT DO NOTHING
                    """, (str(order.id), item.product_id,
                          item.product_name, item.quantity, item.unit_price))

    def find_by_id(self, order_id: UUID) -> Order | None:
        with psycopg2.connect(self._conn_str) as conn:
            with conn.cursor() as cur:
                cur.execute("SELECT * FROM orders WHERE id = %s", (str(order_id),))
                row = cur.fetchone()
                if not row:
                    return None
                return self._to_domain(row, cur)

    def find_by_customer(self, customer_id: str) -> list[Order]:
        # 구현 생략
        pass

    def _to_domain(self, row, cursor) -> Order:
        # DB 행 → 도메인 모델 변환
        pass


# adapters/input/rest_controller.py
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from uuid import UUID
from domain.services.order_service import OrderService
from domain.models.order import OrderItem

app = FastAPI()


class CreateOrderRequest(BaseModel):
    customer_id: str
    items: list[dict]


class OrderResponse(BaseModel):
    id: str
    customer_id: str
    status: str
    total_amount: float


def create_rest_controller(order_service: OrderService):

    @app.post("/orders", response_model=OrderResponse)
    async def create_order(request: CreateOrderRequest):
        items = [
            OrderItem(
                product_id=i["product_id"],
                product_name=i["product_name"],
                quantity=i["quantity"],
                unit_price=i["unit_price"]
            )
            for i in request.items
        ]
        order = order_service._create_order(request.customer_id, items)
        return OrderResponse(
            id=str(order.id),
            customer_id=order.customer_id,
            status=order.status.value,
            total_amount=order.total_amount
        )

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

    return app

# config/dependency_injection.py
from domain.services.order_service import OrderService
from adapters.output.postgres_order_repo import PostgresOrderRepository
from adapters.output.stripe_payment import StripePaymentGateway
from adapters.input.rest_controller import create_rest_controller


def bootstrap():
    # Output Adapters
    order_repo = PostgresOrderRepository(
        connection_string="postgresql://user:pass@localhost/orders"
    )
    payment_gateway = StripePaymentGateway(
        api_key="sk_test_xxx"
    )

    # Domain Service (Port 구현)
    order_service = OrderService(
        order_repo=order_repo,
        payment_gateway=payment_gateway
    )

    # Input Adapter
    app = create_rest_controller(order_service)

    return app


# DB를 바꾸고 싶다면?
# PostgresOrderRepository → MongoOrderRepository로 교체만 하면 됨!
# 도메인 코드 변경 없음!

import pytest
from domain.models.order import Order, OrderItem, OrderStatus


class TestOrder:
    def test_add_item(self):
        order = Order(customer_id="C001")
        item = OrderItem("P001", "노트북", 1, 1500000)
        order.add_item(item)
        assert len(order.items) == 1
        assert order.total_amount == 1500000

    def test_cannot_add_item_to_confirmed_order(self):
        order = Order(customer_id="C001")
        order.add_item(OrderItem("P001", "노트북", 1, 1500000))
        order.confirm()
        with pytest.raises(ValueError, match="확정된 주문"):
            order.add_item(OrderItem("P002", "마우스", 1, 50000))

    def test_cannot_confirm_empty_order(self):
        order = Order(customer_id="C001")
        with pytest.raises(ValueError, match="상품이 없는"):
            order.confirm()

    def test_cannot_cancel_shipped_order(self):
        order = Order(customer_id="C001")
        order.add_item(OrderItem("P001", "노트북", 1, 1500000))
        order.confirm()
        order.status = OrderStatus.SHIPPED
        with pytest.raises(ValueError, match="배송된"):
            order.cancel()

from unittest.mock import MagicMock
from domain.services.order_service import OrderService
from domain.models.order import Order, OrderItem


class TestOrderService:
    def setup_method(self):
        self.mock_repo = MagicMock()
        self.mock_payment = MagicMock()
        self.service = OrderService(self.mock_repo, self.mock_payment)

    def test_create_order(self):
        items = [OrderItem("P001", "노트북", 1, 1500000)]
        order = self.service._create_order("C001", items)

        assert order.customer_id == "C001"
        assert len(order.items) == 1
        self.mock_repo.save.assert_called_once()

    def test_confirm_order_with_payment(self):
        order = Order(customer_id="C001")
        order.add_item(OrderItem("P001", "노트북", 1, 1500000))
        self.mock_repo.find_by_id.return_value = order
        self.mock_payment.charge.return_value = True

        result = self.service.confirm_order(order.id)

        self.mock_payment.charge.assert_called_once()
        assert result.status.value == "confirmed"

요구사항: 주문을 취소하면 결제를 환불하고 재고를 되돌린다

adapters/input/rest_controller.py       DELETE /orders/{id} 수신        ← 주도 어댑터
ports/input/order_use_case.py           CancelOrderUseCase 계약        ← 주도 포트
domain/services/order_service.py        취소 절차를 조율                ← 유스케이스
domain/models/order.py                  Order.cancel() 규칙 판정        ← 도메인
ports/output/order_repository.py        save / find_by_id              ← 피주도 포트
ports/output/payment_gateway.py         refund                         ← 피주도 포트
ports/output/inventory_port.py          restock  (이번에 새로 추가)      ← 피주도 포트
adapters/output/postgres_order_repo.py  실제 SQL                       ← 피주도 어댑터
adapters/output/stripe_payment.py       실제 Stripe 호출                ← 피주도 어댑터
adapters/output/wms_inventory.py        실제 WMS HTTP 호출              ← 피주도 어댑터

# ports/output/inventory_port.py — 도메인이 인프라에게 요구하는 계약
from abc import ABC, abstractmethod
from uuid import UUID


class InventoryPort(ABC):
    @abstractmethod
    def restock(self, order_id: UUID, lines: list[tuple[str, int]]) -> None:
        """(product_id, quantity) 목록만큼 재고를 되돌린다."""


# domain/models/order.py — 규칙 판정은 오직 여기서
    def cancel(self) -> None:
        if self.status == OrderStatus.SHIPPED:
            raise ValueError("배송된 주문은 취소할 수 없습니다")
        if self.status == OrderStatus.CANCELLED:
            raise ValueError("이미 취소된 주문입니다")
        self.status = OrderStatus.CANCELLED

    def restock_lines(self) -> list[tuple[str, int]]:
        return [(i.product_id, i.quantity) for i in self.items]


# domain/services/order_service.py — 조율만 하고 규칙은 판정하지 않는다
class CancelOrderService(CancelOrderUseCase):
    def __init__(self, orders: OrderRepository,
                 payments: PaymentGateway,
                 inventory: InventoryPort):
        self._orders = orders
        self._payments = payments
        self._inventory = inventory

    def execute(self, order_id: UUID) -> Order:
        order = self._orders.find_by_id(order_id)
        if order is None:
            raise ValueError(f"주문을 찾을 수 없습니다: {order_id}")

        order.cancel()                        # 규칙은 도메인이 판정
        self._orders.save(order)              # 피주도 포트 1
        self._payments.refund(order.id)       # 피주도 포트 2
        self._inventory.restock(order.id, order.restock_lines())  # 피주도 포트 3
        return order

# tests/test_cancel_order.py — 가짜 어댑터만으로 전 구간을 돈다
import pytest
from domain.models.order import Order, OrderItem, OrderStatus
from domain.services.order_service import CancelOrderService


class FakeOrderRepository:
    def __init__(self, order):
        self._order, self.saved = order, []

    def find_by_id(self, order_id):
        return self._order

    def save(self, order):
        self.saved.append(order.status)


class FakePaymentGateway:
    def __init__(self):
        self.refunded = []

    def refund(self, order_id):
        self.refunded.append(order_id)
        return True


class FakeInventory:
    def __init__(self):
        self.restocked = []

    def restock(self, order_id, lines):
        self.restocked.append(lines)


def make_order() -> Order:
    order = Order(customer_id="C001")
    order.add_item(OrderItem("P001", "노트북", 1, 1_500_000))
    order.add_item(OrderItem("P002", "마우스", 2, 50_000))
    return order


def test_cancel_refunds_and_restocks():
    order = make_order()
    repo, pay, inv = FakeOrderRepository(order), FakePaymentGateway(), FakeInventory()
    service = CancelOrderService(repo, pay, inv)

    result = service.execute(order.id)

    assert result.status is OrderStatus.CANCELLED
    assert repo.saved == [OrderStatus.CANCELLED]
    assert pay.refunded == [order.id]
    assert inv.restocked == [[("P001", 1), ("P002", 2)]]


def test_shipped_order_is_not_cancellable():
    order = make_order()
    order.status = OrderStatus.SHIPPED
    service = CancelOrderService(FakeOrderRepository(order),
                                 FakePaymentGateway(), FakeInventory())

    with pytest.raises(ValueError, match="배송된"):
        service.execute(order.id)

$ pytest tests/test_cancel_order.py -q
..                                                            [100%]
2 passed in 0.03s

$ pytest tests/test_cancel_order.py -q      # refund 호출을 지운 뒤
.F                                                            [100%]
=================================== FAILURES ===================================
_____________________ test_cancel_refunds_and_restocks ________________________
E       AssertionError: assert [] == [UUID('9f0c1a3e-...')]
E         Right contains one more item: UUID('9f0c1a3e-...')
tests/test_cancel_order.py:57: AssertionError
1 failed, 1 passed in 0.04s

# ❌ 빈혈: 판정이 유스케이스에 있다
class OrderService:
    def cancel(self, order_id):
        order = self._orders.find_by_id(order_id)
        if order.status == "shipped":          # 규칙이 여기 있다
            raise ValueError("배송된 주문은 취소 불가")
        if order.status == "cancelled":        # 다른 유스케이스에도 또 있다
            raise ValueError("이미 취소됨")
        order.status = "cancelled"             # 상태를 밖에서 바꾼다
        self._orders.save(order)

# ✅ 판정을 엔티티로 옮긴다
class OrderService:
    def cancel(self, order_id):
        order = self._orders.find_by_id(order_id)
        order.cancel()                          # 규칙은 엔티티가 판정
        self._orders.save(order)

# ❌ 새는 포트 — 인프라가 시그니처에 드러난다
class OrderRepository(ABC):
    @abstractmethod
    def find_by_id(self, session, order_id) -> "OrderTable":   # ORM 모델 반환
        ...

    @abstractmethod
    def execute_query(self, sql: str) -> list[tuple]:          # 저장소의 어휘
        ...

# ✅ 도메인 타입과 도메인 어휘만 남긴다
class OrderRepository(ABC):
    @abstractmethod
    def find_by_id(self, order_id: UUID) -> Order | None:
        ...

    @abstractmethod
    def find_by_customer(self, customer_id: str) -> list[Order]:
        ...

# 도메인/포트가 인프라를 import 하는지 확인 — 결과가 비어야 정상
grep -rnE '^[[:space:]]*(from|import)[[:space:]]+(sqlalchemy|psycopg2|pymongo|django|fastapi|redis|boto3)' \
     domain/ ports/

# CI에 그대로 넣을 수 있는 형태 (하나라도 걸리면 실패)
if grep -rqE '^[[:space:]]*(from|import)[[:space:]]+(sqlalchemy|psycopg2|django|fastapi)' domain/ ports/; then
  echo "FAIL: 도메인 또는 포트가 인프라를 import 합니다"
  exit 1
fi

# ports/output/unit_of_work.py — 도메인 어휘로 감싼 트랜잭션 경계
from abc import ABC, abstractmethod


class UnitOfWork(ABC):
    """하나의 작업 단위라는 개념만 노출한다.
    커밋과 롤백이 실제로 무엇인지는 어댑터만 안다."""

    @abstractmethod
    def __enter__(self) -> "UnitOfWork":
        ...

    @abstractmethod
    def __exit__(self, exc_type, exc, tb) -> None:
        ...

    @abstractmethod
    def commit(self) -> None:
        ...


# 유스케이스에서
    def execute(self, order_id: UUID) -> Order:
        with self._uow:
            order = self._orders.find_by_id(order_id)
            order.cancel()
            self._orders.save(order)
            self._inventory.restock(order.id, order.restock_lines())
            self._uow.commit()

        # 결제 환불은 같은 트랜잭션에 넣을 수 없다 (외부 시스템)
        # → 아웃박스에 기록하고 별도 워커가 재시도한다
        self._outbox.append("order.cancelled", order.id)
        return order