Skip to content
←

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

✨ Learn with Quiz
|

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

Hexagonal Architecture

들어가며

"데이터베이스를 MySQL에서 PostgreSQL로 바꿔야 합니다." "REST API 대신 gRPC도 지원해야 합니다."

이런 요구사항을 들었을 때, 코드 전체를 뒤집어야 한다면 아키텍처에 문제가 있는 것입니다. **Hexagonal Architecture(헥사고날 아키텍처)**는 비즈니스 로직을 외부 의존성에서 완벽히 분리하여 이런 변경을 쉽게 만듭니다.

핵심 개념

3가지 핵심 구성 요소

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와의 차이

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

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

Python 실전 구현

프로젝트 구조

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

도메인 서비스 (Use Case 구현)

# 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()

Mock을 사용한 서비스 테스트

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"

마무리

Hexagonal Architecture의 핵심 가치:

  1. 도메인 독립성: 비즈니스 로직이 DB, 프레임워크에 종속되지 않음
  2. 교체 용이성: 어댑터만 바꾸면 외부 시스템 변경 가능
  3. 테스트 용이성: 도메인은 Mock 없이, 서비스는 포트 Mock으로 테스트
  4. 포트가 계약: 인터페이스(Port)가 내부와 외부의 명확한 계약 역할
📝 퀴즈 (6문제)

Q1. Hexagonal Architecture의 세 가지 핵심 구성 요소는? Domain(Core), Port(인터페이스), Adapter(구현)

Q2. Input Port와 Output Port의 차이는? Input Port: 외부에서 도메인으로 (Use Case), Output Port: 도메인에서 외부로 (Repository 인터페이스)

Q3. Hexagonal Architecture에서 의존성 방향은? 바깥(Adapter) → 안쪽(Domain). 도메인은 외부에 의존하지 않음

Q4. DB를 MySQL에서 PostgreSQL로 바꿀 때 수정해야 하는 것은? Output Adapter만 교체 (도메인 코드 변경 없음)

Q5. 도메인 단위 테스트에서 Mock이 불필요한 이유는? 도메인 모델에 외부 의존성이 없으므로 순수 로직만 테스트 가능

Q6. Layered Architecture 대비 Hexagonal의 주요 장점은? 비즈니스 로직이 DB/프레임워크에 종속되지 않아 교체와 테스트가 용이

Hexagonal Architecture (Ports & Adapters) Practical Guide — The Core of Clean Architecture

Hexagonal Architecture

Introduction

"We need to switch the database from MySQL to PostgreSQL." "We need to support gRPC in addition to the REST API."

If hearing requirements like these means you have to rewrite the entire codebase, there is a problem with your architecture. Hexagonal Architecture completely decouples business logic from external dependencies, making such changes easy.

Core Concepts

3 Key Components

1. Domain (Core)
   - The heart of business logic
   - No external dependencies (pure code)
   - Entity, Value Object, Domain Service

2. Port
   - Interface (contract) between domain and the outside world
   - Input Port: Outside to Domain (Use Case)
   - Output Port: Domain to Outside (Repository interface)

3. Adapter
   - Concrete implementation of a Port
   - Input Adapter: REST Controller, gRPC Handler, CLI
   - Output Adapter: MySQL Repository, Redis Cache, HTTP Client

Difference from Layered Architecture

# Layered Architecture (Traditional)
# Controller → Service → Repository → DB
# Dependency direction: Top → Bottom (coupled to DB)

# Hexagonal Architecture
# Adapter → Port → Domain ← Port ← Adapter
# Dependency direction: Outside → Inside (Domain is the center)

Hands-On Implementation with Python

Project Structure

order-service/
├── domain/                    # Core Domain
│   ├── models/
│   │   ├── order.py          # Entity
│   │   └── order_item.py     # Value Object
│   └── services/
│       └── order_service.py  # Domain Service
├── ports/                     # Ports (Interfaces)
│   ├── input/
│   │   └── order_use_case.py # Input Port
│   └── output/
│       ├── order_repository.py    # Output Port
│       └── payment_gateway.py     # Output Port
├── adapters/                  # Adapters (Implementations)
│   ├── input/
│   │   ├── rest_controller.py     # REST API
│   │   └── grpc_handler.py        # gRPC
│   └── output/
│       ├── postgres_order_repo.py # PostgreSQL implementation
│       ├── redis_order_cache.py   # Redis cache
│       └── stripe_payment.py      # Stripe payment
└── config/
    └── dependency_injection.py    # DI configuration

Domain Model

# 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:
    """Order Entity — contains business rules"""
    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("Cannot add items to a confirmed order")
        if item.quantity <= 0:
            raise ValueError("Quantity must be at least 1")
        self.items.append(item)

    def confirm(self) -> None:
        if not self.items:
            raise ValueError("Cannot confirm an order with no items")
        if self.status != OrderStatus.PENDING:
            raise ValueError(f"Cannot confirm from '{self.status.value}' status")
        self.status = OrderStatus.CONFIRMED

    def cancel(self) -> None:
        if self.status == OrderStatus.SHIPPED:
            raise ValueError("Cannot cancel a shipped order")
        self.status = OrderStatus.CANCELLED

Port Definitions

# 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 Service (Use Case Implementation)

# 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):
    """Order Service — Input Port implementation"""

    def __init__(
        self,
        order_repo: OrderRepository,
        payment_gateway: PaymentGateway
    ):
        # Depends on Output Port (not the concrete implementation!)
        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 not found: {order_id}")

        # Process payment
        success = self._payment_gateway.charge(
            order_id=order.id,
            amount=order.total_amount,
            customer_id=order.customer_id
        )
        if not success:
            raise ValueError("Payment failed")

        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 not found: {order_id}")

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

Adapter Implementation

# 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]:
        # Implementation omitted
        pass

    def _to_domain(self, row, cursor) -> Order:
        # Convert DB row to domain model
        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

Dependency Injection Configuration

# 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 implementation)
    order_service = OrderService(
        order_repo=order_repo,
        payment_gateway=payment_gateway
    )

    # Input Adapter
    app = create_rest_controller(order_service)

    return app


# Want to switch databases?
# Just replace PostgresOrderRepository with MongoOrderRepository!
# No changes to domain code!

Testing Strategy

Domain Unit Tests (No External Dependencies)

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", "Laptop", 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", "Laptop", 1, 1500000))
        order.confirm()
        with pytest.raises(ValueError, match="confirmed order"):
            order.add_item(OrderItem("P002", "Mouse", 1, 50000))

    def test_cannot_confirm_empty_order(self):
        order = Order(customer_id="C001")
        with pytest.raises(ValueError, match="no items"):
            order.confirm()

    def test_cannot_cancel_shipped_order(self):
        order = Order(customer_id="C001")
        order.add_item(OrderItem("P001", "Laptop", 1, 1500000))
        order.confirm()
        order.status = OrderStatus.SHIPPED
        with pytest.raises(ValueError, match="shipped"):
            order.cancel()

Service Tests with Mocks

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", "Laptop", 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", "Laptop", 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"

Conclusion

The core values of Hexagonal Architecture:

  1. Domain Independence: Business logic is not coupled to DB or frameworks
  2. Easy Replacement: Just swap adapters to change external systems
  3. Testability: Domain tests need no mocks; service tests only mock ports
  4. Ports as Contracts: Interfaces (Ports) serve as clear contracts between internal and external boundaries
Quiz (6 Questions)

Q1. What are the three key components of Hexagonal Architecture? Domain (Core), Port (Interface), Adapter (Implementation)

Q2. What is the difference between Input Port and Output Port? Input Port: From outside to domain (Use Case). Output Port: From domain to outside (Repository interface)

Q3. What is the direction of dependencies in Hexagonal Architecture? Outside (Adapter) to inside (Domain). The domain does not depend on anything external.

Q4. What needs to change when switching the DB from MySQL to PostgreSQL? Only the Output Adapter needs to be replaced (no changes to domain code)

Q5. Why are mocks unnecessary in domain unit tests? Because domain models have no external dependencies, so only pure logic is tested

Q6. What is the main advantage of Hexagonal over Layered Architecture? Business logic is not coupled to DB/frameworks, making replacement and testing easier

Quiz

Q1: What is the main topic covered in "Hexagonal Architecture (Ports & Adapters) Practical Guide — The Core of Clean Architecture"?

From the core concepts of Hexagonal Architecture (Ports & Adapters) to hands-on implementation with Python/Spring Boot and testing strategies. Learn how to completely decouple business logic from external dependencies.

Q2: What is Core Concepts?3 Key Components Difference from Layered Architecture

Q3: Explain the core concept of Hands-On Implementation with Python. Project Structure Domain Model Port Definitions Domain Service (Use Case Implementation) Adapter Implementation Dependency Injection Configuration

Q4: What are the key aspects of Testing Strategy? Domain Unit Tests (No External Dependencies) Service Tests with Mocks