Cosmic Python 제1부(1): 도메인 모델링, Repository 패턴, 그리고 추상화

제1부: 도메인 모델링을 지원하는 아키텍처 구축

“대부분의 개발자는 도메인 모델을 본 적이 없다. 그들이 본 것은 데이터 모델뿐이다.” — Cyrille Martraire, DDD EU 2017

---

왜 이 책을 읽어야 할까?

우리가 흔히 하는 개발 방식을 떠올려보자.

1. 데이터베이스 스키마를 먼저 설계하고
2. ORM 모델을 만들고
3. 그 위에 비즈니스 로직을 얹는다

이 방식이 뭐가 문제일까?

비즈니스 로직이 코드 여기저기에 흩어진다. 컨트롤러에도 있고, 모델에도 있고, 유틸리티 함수에도 있다. 시간이 지나면 “이 로직이 어디에 있더라?” 하면서 코드를 뒤지게 된다.

Cosmic Python의 저자들은 이 접근을 뒤집자고 제안한다.

“저장소가 행동을 결정하는 게 아니라, 행동이 저장소를 결정해야 한다.”

비즈니스 로직을 먼저 설계하고, 데이터베이스는 나중에 생각하자는 것이다.

---

Part 1에서 다루는 핵심 패턴들

Part 1은 총 7개의 챕터로 구성되어 있다. 각 챕터가 하나의 패턴을 점진적으로 쌓아 올린다.

챕터	패턴	핵심 질문
Ch1	도메인 모델링	“비즈니스 로직을 어떻게 코드로 표현할까?”
Ch2	Repository 패턴	“도메인 모델을 DB에서 어떻게 분리할까?”
Ch3	결합도와 추상화	“좋은 추상화란 무엇일까?”
Ch4	Service Layer	“유스케이스를 어디에 넣을까?”
Ch5	TDD	“테스트 피라미드를 어떻게 최적화할까?”
Ch6	Unit of Work	“트랜잭션을 어떻게 깔끔하게 관리할까?”
Ch7	Aggregate	“일관성 경계를 어떻게 설정할까?”

---

Chapter 1: 도메인 모델링(Domain Modeling)

도메인 모델이란?

도메인 모델은 “비즈니스 문제를 코드로 표현한 것”이다. 이 책에서는 가구 소매업체의 재고 할당 시스템을 예시로 사용한다.

핵심 비즈니스 규칙은 이렇다.

• 고객이 주문하면 → 가용 재고에서 할당(allocate)한다
• 창고 재고를 먼저 할당하고, 입고 예정 배치는 나중에 할당한다
• 같은 주문 라인을 두 번 할당할 수 없다
• 재고가 부족하면 할당 실패

---

값 객체(Value Object)와 엔터티(Entity)

DDD에서 가장 기본적인 두 가지 개념이다.

값 객체: “무엇이냐”가 중요

@dataclass(frozen=True)
class OrderLine:
    orderid: str
    sku: str
    qty: int

frozen=True로 불변(immutable)하게 만든다. 같은 orderid, sku, qty를 가진 두 OrderLine은 동일한 것이다. 정체성(identity)이 없고, 값 자체가 의미다.

엔터티: “누구냐”가 중요

class Batch:
    def __init__(self, ref: str, sku: str, qty: int, eta: Optional[date]):
        self.reference = ref
        self.sku = sku
        self.eta = eta
        self._purchased_quantity = qty
        self._allocations = set()

    def allocate(self, line: OrderLine):
        if self.can_allocate(line):
            self._allocations.add(line)

    def can_allocate(self, line: OrderLine) -> bool:
        return self.sku == line.sku and self.available_quantity >= line.qty

    @property
    def available_quantity(self) -> int:
        return self._purchased_quantity - self.allocated_quantity

    @property
    def allocated_quantity(self) -> int:
        return sum(line.qty for line in self._allocations)

Batch는 reference라는 고유 식별자를 가진다. 수량이 바뀌거나 할당이 추가되어도, 같은 reference면 같은 Batch다.

구분	값 객체(Value Object)	엔터티(Entity)
동일성 기준	값이 같으면 같다	식별자가 같으면 같다
불변성	불변	변경 가능
예시	OrderLine, 주소, 가격	Batch, 주문, 고객

---

도메인 서비스 함수

모든 로직이 객체에 담기는 건 아니다. “할당한다”라는 행위는 특정 객체에 속하지 않는다. 이런 경우 도메인 서비스 함수로 표현한다.

def allocate(line: OrderLine, batches: List[Batch]) -> str:
    try:
        batch = next(
            b for b in sorted(batches) if b.can_allocate(line)
        )
    except StopIteration:
        raise OutOfStock(f"Out of stock for sku {line.sku}")
    batch.allocate(line)
    return batch.reference

비즈니스 규칙이 그대로 코드에 녹아 있다.

• sorted(batches): 창고 재고(ETA 없음)를 먼저, 그 다음은 입고일 빠른 순서로
• can_allocate: SKU가 맞고 수량이 충분한지 확인
• OutOfStock: 재고 부족이라는 도메인 예외

---

Chapter 2: Repository 패턴

문제: ORM이 도메인을 오염시킨다

보통 Django나 SQLAlchemy를 쓰면, 도메인 모델이 ORM에 종속된다. models.Model을 상속받거나, ORM 필드를 직접 정의하게 되면 도메인 모델이 인프라를 알게 된다.

그런데 도메인 모델은 비즈니스 로직만 알아야 한다. 데이터베이스가 MySQL인지 PostgreSQL인지, ORM이 뭔지 몰라도 되어야 한다.

해결: 의존성 역전(Dependency Inversion)

“인프라가 도메인에 의존해야지, 도메인이 인프라에 의존하면 안 된다.”

Repository 패턴은 영속성 저장소에 대한 추상화다. 마치 인메모리 컬렉션처럼 add()와 get()만 제공한다.

class AbstractRepository(abc.ABC):
    @abc.abstractmethod
    def add(self, entity):
        raise NotImplementedError

    @abc.abstractmethod
    def get(self, reference) -> model.Batch:
        raise NotImplementedError

실제 구현 (SQLAlchemy)

class SqlAlchemyRepository(AbstractRepository):
    def __init__(self, session):
        self.session = session

    def add(self, batch):
        self.session.add(batch)

    def get(self, reference):
        return self.session.query(model.Batch)\
            .filter_by(reference=reference).one()

테스트용 Fake 구현

class FakeRepository(AbstractRepository):
    def __init__(self, batches):
        self._batches = set(batches)

    def add(self, batch):
        self._batches.add(batch)

    def get(self, reference):
        return next(b for b in self._batches
                    if b.reference == reference)

데이터베이스 없이도 테스트를 돌릴 수 있다. 테스트가 빨라지고 안정적이 된다.

AbstractRepository는 포트(Port), SqlAlchemyRepository와 FakeRepository는 어댑터(Adapter)다. 이것이 바로 포트-어댑터(헥사고날) 아키텍처의 핵심이다.

---

Chapter 3: 결합도와 추상화

“테스트 가능하게 설계한다는 것은 곧, 확장 가능하게 설계한다는 뜻이다.”

문제: 비즈니스 로직과 I/O가 뒤엉켜 있다

파일 동기화 도구를 예로 들어보자. 소스 디렉토리와 목적지 디렉토리를 비교해서 파일을 복사/이동/삭제하는 프로그램이다.

def sync(source, dest):
    # 1. 소스 파일 해시 수집 (I/O)
    source_hashes = {}
    for folder, _, files in os.walk(source):
        for fn in files:
            source_hashes[hash_file(Path(folder) / fn)] = fn

    seen = set()

    # 2. 목적지 파일과 비교 + 즉시 실행 (로직 + I/O 혼합)
    for folder, _, files in os.walk(dest):
        for fn in files:
            dest_path = Path(folder) / fn
            dest_hash = hash_file(dest_path)
            seen.add(dest_hash)

            if dest_hash not in source_hashes:
                dest_path.remove()
            elif fn != source_hashes[dest_hash]:
                shutil.move(dest_path, Path(folder) / source_hashes[dest_hash])

    for source_hash, fn in source_hashes.items():
        if source_hash not in seen:
            shutil.copy(Path(source) / fn, Path(dest) / fn)

이 코드의 문제점:

• 비즈니스 로직(어떤 파일을 복사/이동/삭제할지 결정)과 I/O(실제 파일 조작)가 한 함수에 섞여 있다
• 테스트하려면 실제 파일 시스템이 필요하다 → 느리고 불안정하다
• --dry-run 모드나 클라우드 스토리지 지원을 추가하기 어렵다

이것이 바로 결합(coupling)이다. 컴포넌트 A를 변경하면 컴포넌트 B가 깨질까 두려운 상태.

해결: 세 가지 책임을 분리하자

하나의 함수에 묻혀 있던 세 가지 책임을 명확히 분리한다.

1단계: I/O 수집 (Imperative Shell)

def read_paths_and_hashes(root):
    hashes = {}
    for folder, _, files in os.walk(root):
        for fn in files:
            hashes[hash_file(Path(folder) / fn)] = fn
    return hashes

2단계: 순수 비즈니스 로직 (Functional Core)

def determine_actions(source_hashes, dest_hashes, source_folder, dest_folder):
    for sha, filename in source_hashes.items():
        if sha not in dest_hashes:
            sourcepath = Path(source_folder) / filename
            destpath = Path(dest_folder) / filename
            yield "COPY", sourcepath, destpath

        elif dest_hashes[sha] != filename:
            olddestpath = Path(dest_folder) / dest_hashes[sha]
            newdestpath = Path(dest_folder) / filename
            yield "MOVE", olddestpath, newdestpath

    for sha, filename in dest_hashes.items():
        if sha not in source_hashes:
            yield "DELETE", dest_folder / filename

I/O가 전혀 없다. 딕셔너리를 받아서 튜플을 반환할 뿐이다. “무엇을 할지”를 결정하지만, “어떻게 실행할지”는 모른다.

3단계: 오케스트레이션 (Imperative Shell)

def sync(source, dest):
    source_hashes = read_paths_and_hashes(source)
    dest_hashes = read_paths_and_hashes(dest)

    actions = determine_actions(source_hashes, dest_hashes, source, dest)

    for action, *paths in actions:
        if action == "COPY":
            shutil.copyfile(*paths)
        if action == "MOVE":
            shutil.move(*paths)
        if action == "DELETE":
            os.remove(paths[0])

이것이 Gary Bernhardt의 Functional Core, Imperative Shell 패턴이다.

• Functional Core: 순수 함수로 비즈니스 로직 수행 (side-effect 없음)
• Imperative Shell: 바깥에서 I/O를 처리하고 결과를 전달

테스트가 깔끔해진다

def test_when_a_file_exists_in_the_source_but_not_the_destination():
    source_hashes = {"hash1": "fn1"}
    dest_hashes = {}
    actions = determine_actions(source_hashes, dest_hashes, Path("/src"), Path("/dst"))
    assert list(actions) == [("COPY", Path("/src/fn1"), Path("/dst/fn1"))]


def test_when_a_file_has_been_renamed_in_the_source():
    source_hashes = {"hash1": "fn1"}
    dest_hashes = {"hash1": "fn2"}
    actions = determine_actions(source_hashes, dest_hashes, Path("/src"), Path("/dst"))
    assert list(actions) == [("MOVE", Path("/dst/fn2"), Path("/dst/fn1"))]

파일 시스템이 필요 없다. 딕셔너리 → 리스트 변환을 검증할 뿐이다.

---

Fake와 의존성 주입으로 Edge-to-Edge 테스트

순수 함수 테스트만으로는 전체 흐름을 검증하기 어렵다. 의존성 주입(DI)을 사용하면 최상위 sync() 함수도 테스트할 수 있다.

class FakeFilesystem:
    def __init__(self, path_hashes):
        self.path_hashes = path_hashes
        self.actions = []

    def read(self, path):
        return self.path_hashes[path]

    def copy(self, source, dest):
        self.actions.append(("COPY", source, dest))

    def move(self, source, dest):
        self.actions.append(("MOVE", source, dest))

    def delete(self, dest):
        self.actions.append(("DELETE", dest))

이 Fake는 실제 파일 시스템을 흉내내면서, 어떤 동작이 수행되었는지 기록한다. sync(source, dest, filesystem=FakeFilesystem(...))처럼 주입하면 된다.

DI의 실질적 이점은 테스트뿐만이 아니다. --dry-run 플래그, FTP/S3 스토리지 전환 등 확장성도 함께 얻는다.

---

Mock vs Fake: 어떤 걸 써야 할까?

저자들은 unittest.mock.patch를 피하는 이유를 세 가지로 든다.

1. 설계를 개선하지 않는다: shutil.copy를 mock할 수는 있지만, --dry-run이나 FTP 지원은 여전히 불가능하다
2. 구현에 결합된다: assert_called_once_with로 “어떻게 호출되었는지”를 검증하면, 내부 리팩터링 시 테스트가 깨진다
3. 테스트가 복잡해진다: setup 코드가 테스트의 의도를 가린다

구분	Mock	Fake
검증 대상	호출 방식 (behavior)	최종 상태 (state)
스타일	London-school TDD	Classic-style TDD
장점	세밀한 행위 검증	리팩터링에 강하다
단점	구현 변경 시 테스트 깨짐	Fake 구현 비용

이 책은 Classic-style(Fake 기반)을 따른다.

---

좋은 추상화를 찾는 체크리스트

• 지저분한 시스템의 상태를 익숙한 파이썬 자료구조로 표현할 수 있는가?
• “무엇을 할지”와 “어떻게 실행할지”를 분리할 수 있는가?
• 어디에 이음새(seam)를 넣어 추상화를 끼워 넣을 수 있는가?
• 암묵적인 개념을 명시적으로 만들 수 있는가?

“연습하면 덜 불완전해진다!” (Practice makes less imperfect!)

---

다음 글: 제1부(2): Service Layer와 TDD 전략에서 이어진다.

---

참고 자료

• Cosmic Python - Part 1: Building an Architecture to Support Domain Modeling
• Cosmic Python - Chapter 1: Domain Modeling
• Cosmic Python - Chapter 2: Repository Pattern
• Cosmic Python - Chapter 3: A Brief Interlude: On Coupling and Abstractions
• Cosmic Python 전체 목차