Architecture Deep Dive

헥사고날 아키텍처

Ports & Adapters — 도메인을 외부 세계로부터 격리하는 설계

1. 한 줄 정의

도메인(비즈니스 로직)을 중앙에 두고, 외부 세계와의 모든 통신을 port(인터페이스)adapter(구현)로 분리하는 아키텍처.

— Alistair Cockburn, 2005

이름의 유래는 다이어그램을 그렸을 때 도메인 둘레가 육각형(hexagon)으로 보였기 때문. 사실 면의 개수는 의미 없고, "외부와 닿는 면이 여러 개"라는 점이 핵심.

헥사고날 아키텍처 다이어그램 Domain (use cases + entities) HTTP API (driving) CLI Worker DB (driven) LLM / HTTP Cache / Redis
도메인을 둘러싼 6개의 port — 위쪽(driving)에서 들어와 아래쪽(driven)으로 나간다

2. 왜 쓰는가

🧪 테스트 용이성

도메인이 외부 의존성을 모름. fake adapter 만 끼우면 DB·HTTP·LLM 없이 비즈니스 로직 단위 테스트 가능.

🔌 교체 가능성

PostgreSQL → MongoDB, OpenAI → Gemini, REST → GraphQL. 도메인 코드는 한 줄도 안 바뀜.

🛡️ 의존성 역전

외부 SDK 가 도메인을 의존(adapter implements port). 도메인이 SDK 를 의존하지 않음.

📦 도메인 격리

비즈니스 규칙이 프레임워크 변경(FastAPI → Litestar)에 영향 받지 않음. 도메인은 순수 Python/TS.

🧠 인지 부하 감소

"이 코드는 비즈니스 룰인가, 인프라인가?"가 폴더만 봐도 명확. 신규 합류자 온보딩 가속.

⏳ 변경 비용 분리

인프라는 자주 바뀌고(라이브러리 업데이트, 클라우드 이전), 도메인은 천천히 바뀜. 이 두 속도를 분리.

3. 3계층 구조

① Domain (코어)

비즈니스 규칙, 엔티티, use case

  • Entity — 도메인 모델 (Pydantic / dataclass, 외부 의존 0)
  • Port — 외부와의 contract (ABC 인터페이스)
  • Use Case — 비즈니스 로직 (port 만 의존)

⚠️ 외부 SDK·DB session·HTTP client import 금지

↑ implements

② Adapter (구현)

port 의 구체 구현 — 외부 세계의 번역기

  • Driven adapter — DB, LLM, Redis, S3, 외부 API 호출 (도메인이 부르는 쪽)
  • Driving adapter — HTTP route, CLI, message consumer (도메인을 부르는 쪽)

SQLAlchemy, httpx, google.genai 등 실제 라이브러리는 여기서만

↑ wires

③ Composition Root (진입점)

adapter ↔ use case wiring

  • API — FastAPI route (Pydantic schema → use case 호출)
  • Workers — Dramatiq actor (백그라운드 작업)

여기서만 의존성 주입이 일어남. 도메인은 자기 자신을 wire 하지 않음

4. 호출 흐름 — "매물 등록" 시나리오

  1. HTTP requestPOST /properties 도착
  2. API route (driving adapter) 가 Pydantic schema 로 검증 → RegisterPropertyUseCase 호출
  3. Use casePropertyRepository port 의 .save() 호출 — 구현은 모름
  4. SqlAlchemyPropertyRepository (driven adapter) 가 실제 PostgreSQL 에 INSERT
  5. Use case 가 EmbeddingPort.embed() 호출 → VertexAiEmbeddingAdapter 가 LLM 호출 → Qdrant 저장
  6. Use case 가 NotificationPort.publish() 호출 → RedisPubSubAdapter 가 Socket.io 로 broadcast
  7. API route 가 결과를 JSON 으로 응답

use case 코드 어디에도 sqlalchemy, google.genai, redis 라는 단어가 등장하지 않는다. 그것이 헥사고날의 증거.

5. 코드로 보기 — 매물 등록 use case

5-1. Port (도메인이 정의하는 인터페이스)

# domains/properties/ports/property_repository.py
from abc import ABC, abstractmethod
from domains.properties.entities import Property

class PropertyRepository(ABC):
    @abstractmethod
    async def save(self, property: Property) -> Property: ...

    @abstractmethod
    async def find_by_id(self, id: str) -> Property | None: ...

5-2. Use Case (port 만 의존)

# domains/properties/use_cases/register_property.py
from dataclasses import dataclass
from domains.properties.ports import PropertyRepository, EmbeddingPort
from domains.properties.entities import Property

@dataclass
class RegisterPropertyUseCase:
    repo: PropertyRepository       # ← 인터페이스만 알고 있음
    embedder: EmbeddingPort

    async def execute(self, cmd: RegisterPropertyCommand) -> Property:
        prop = Property.create(cmd.address, cmd.price, cmd.area)
        saved = await self.repo.save(prop)
        await self.embedder.embed(saved.id, saved.description)
        return saved

5-3. Adapter (port 의 구체 구현)

# adapters/database/sqlalchemy_property_repository.py
from sqlalchemy.ext.asyncio import AsyncSession
from domains.properties.ports import PropertyRepository
from domains.properties.entities import Property

class SqlAlchemyPropertyRepository(PropertyRepository):
    def __init__(self, session: AsyncSession):
        self.session = session

    async def save(self, property: Property) -> Property:
        row = PropertyORM.from_entity(property)
        self.session.add(row)
        await self.session.flush()
        return row.to_entity()

5-4. Composition Root (route 에서 wiring)

# api/properties.py
from fastapi import APIRouter, Depends

router = APIRouter()

@router.post("/properties")
async def register(
    body: RegisterPropertySchema,
    session: AsyncSession = Depends(get_db),
    embedder: EmbeddingPort = Depends(get_embedder),
):
    use_case = RegisterPropertyUseCase(
        repo=SqlAlchemyPropertyRepository(session),  # ← 여기서만 SQLAlchemy 등장
        embedder=embedder,
    )
    result = await use_case.execute(body.to_command())
    return PropertyResponse.from_entity(result)

5-5. Test — fake adapter 로 격리

# tests/unit/test_register_property.py
class FakePropertyRepo(PropertyRepository):
    def __init__(self):
        self.items = []
    async def save(self, p):
        self.items.append(p); return p
    async def find_by_id(self, id):
        return next((x for x in self.items if x.id == id), None)

async def test_register():
    repo, embedder = FakePropertyRepo(), FakeEmbedder()
    uc = RegisterPropertyUseCase(repo, embedder)
    result = await uc.execute(RegisterPropertyCommand(...))
    assert result.id in [p.id for p in repo.items]
    # DB·LLM 컨테이너 없이 비즈니스 로직만 검증

6. 어디살지 프로젝트에 어떻게 적용됐는가

backend/src/app/ 의 실제 구조:

domains/{domain}/         # 38개 도메인 (auth, properties, contracts, ai, …)
├── entities/              # 순수 모델 (외부 의존 0)
├── ports/                 # ABC 인터페이스
└── use_cases/             # 비즈니스 로직 (port 만 의존)

core/
├── ports/                 # 공용 port (Repository, UoW, Clock, TokenService, …)
├── result.py              # Result[T, E] 모나드 — exception 대신 명시적 에러
└── errors.py              # DomainError 계층

adapters/                  # port 의 구체 구현
├── database/              # SQLAlchemy 2.0 async repo
├── infra/                 # Qdrant, Redis, S3, Dramatiq
├── oauth/                 # Google, Kakao, Naver
├── auth/                  # PasswordHasher, TokenService
└── realtime/              # Socket.io 어댑터

api/                       # FastAPI route (composition root)
workers/                   # Dramatiq actors (composition root)
SSOT (Single Source of Truth) 매핑
데이터 성격SSOT 위치
영속·다중 사용자 공유PostgreSQL (SQLAlchemy + Alembic)
휘발·짧은 TTLRedis
Job queue 상태Dramatiq + Redis broker
Vector embeddingQdrant
서버 상태 클라 캐시TanStack Query (mutate 금지)
클라 ephemeral UIJotai

7. 규칙·금지

✅ DO

  • 새 외부 의존성은 반드시 port 먼저 정의, adapter 는 나중
  • api/·workers/ 가 유일한 composition root
  • 도메인 함수 시그니처에는 port 타입만 등장
  • use case 는 한 가지 비즈니스 행위만 (Single Responsibility)
  • 실패는 Result[T, E] 로 — 도메인에서 exception raise 자제

❌ DON'T

  • 도메인에서 app.adapters.* / sqlalchemy / httpx import
  • API route 에 비즈니스 로직 작성 (route 는 변환·위임만)
  • adapter 가 다른 adapter 를 직접 호출 (도메인을 통해서만)
  • port 에 구현 디테일 누출 (SqlAlchemyRow 반환 타입 등)
  • services/ 신규 추가 — cross-domain 외엔 use_cases/

8. 다른 아키텍처와 비교

스타일의존 방향핵심 차이
Layered (3-tier) UI → Service → DAO → DB 도메인이 DB 를 직접 의존 → 인프라 변경이 도메인까지 침투
Hexagonal UI → UseCase → Port ← Adapter ← DB 의존성 역전. 도메인은 port 만 알고, adapter 가 port 를 구현
Onion UI → App → Domain ← Infra 헥사고날의 동심원 표현. 본질적으로 동일
Clean Architecture Frameworks → Adapters → UseCases → Entities 헥사고날의 4계층 변형. Uncle Bob 의 재포장

Hexagonal · Onion · Clean 셋은 본질이 같다 — 의존성 역전을 통한 도메인 보호. 다이어그램이 다를 뿐.

9. FAQ

Q. Port 와 Repository pattern 의 차이?

Repository 는 port 의 한 종류. Port 는 더 넓은 개념 — DB 외에도 LLM, 메시지 브로커, 외부 API 등 모든 외부 통신이 port. Repository 는 "영속성"이라는 특정 port 의 관용 명칭.

Q. CRUD 만 있는 작은 앱에도 헥사고날이 필요한가?

아니. 도메인 로직이 거의 없으면 오버엔지니어링. 헥사고날의 효용은 비즈니스 규칙이 복잡할수록 커진다. 어디살지처럼 38개 도메인이면 필수.

Q. ORM 모델 = 도메인 엔티티 로 써도 되는가?

안 됨. ORM 모델은 SQLAlchemy 의존 → 도메인 오염. PropertyORM.to_entity() / Property.to_orm() 변환 메서드로 분리. 처음엔 귀찮지만 ORM 교체 시 진가 발휘.

Q. 트랜잭션 경계는 어디?

Use case 단위. UnitOfWork port 를 도입해 use case 내에서 async with uow: 로 묶음. 어디살지 core/ports/uow.py 참조.

Q. 의존성 주입 프레임워크(dependency-injector 등)를 써야 하나?

FastAPI 의 Depends 로 충분. 별도 DI 컨테이너는 어지간히 큰 프로젝트 아니면 과잉. 어디살지도 FastAPI native + 함수형 wiring 으로 운영.