d01 - 6. 데이터 처리 로직 구현

6교시 · 데이터 처리 로직 구현

1일차 16:00–17:00 · 이론/실습 선행: 5교시(SalesRecord), 4교시(SalesRepository 포트)

---

🎯 학습 목표

• CSV에서 매출 레코드를 적재하는 어댑터를 TDD로 구현한다.
• 잘못된 행을 분석을 멈추지 않고 "제외 사유와 함께 건너뛰는" 전처리 규칙을 만든다.
• 어댑터가 유스케이스의 포트(SalesRepository) 를 구현하도록 한다.

---

📖 개념 1 — 어댑터의 역할: 더러운 바깥을 깨끗한 도메인으로 통역

현실의 CSV는 깨끗하지 않습니다. 빈 칸, 잘못된 날짜 형식, 음수 수량이 섞여 있습니다. 반면 도메인(SalesRecord)은 깨끗한 값만 받습니다(5교시에서 검증 규칙을 넣었죠). 이 둘 사이를 잇는 통역사가 어댑터입니다.

아래는 데이터 흐름 도식입니다(화살표 = 데이터가 가는 방향).

[지저분한 CSV 파일]
        │  (행마다 검증·변환 시도)
        ▼
[ CsvSalesRepository (어댑터) ]
        │
        ├──(검증 통과)──▶  list[SalesRecord]   (깨끗한 도메인 객체)
        │
        └──(검증 실패)──▶  skipped 목록         (제외된 행 + 사유)
                            (버리지 않고 보고)

 

🔍 관찰 포인트 — 어댑터는 pandas를 써도 되지만, 내보내는 것은 도메인 객체입니다. 어댑터는 3계층이므로 pandas를 사용해도 됩니다(CSV를 읽는 데 편리). 하지만 결과로 바깥에 내보내는 것은 pandas의 DataFrame이 아니라 list[SalesRecord](도메인 객체) 입니다. 만약 DataFrame을 그대로 유스케이스로 넘기면, pandas가 도메인 안쪽까지 스며들어 의존성 규칙이 무너집니다. 어댑터의 책임은 "외부 형식을 도메인 형식으로 번역해 차단막 역할을 하는 것" 입니다.

📖 개념 2 — "멈추지 않고 건너뛰기"가 왜 중요한가

데이터 1000행 중 3행이 잘못됐다고 전체 분석을 중단하면, 사용자는 답답합니다. 좋은 데이터 제품은 다음처럼 동작합니다.

• 잘못된 행은 분석에서 제외하되,
• 무엇을, 왜 제외했는지 사용자에게 보고합니다(예: "6행: quantity must be > 0").

이렇게 하면 사용자는 "997행으로 분석된 결과 + 제외된 3행의 사유"를 함께 받아, 신뢰하고 사용할 수 있습니다. 이 동작을 이번 교시에 TDD로 구현합니다.

---

🔴 Red — 적재 동작의 약속을 테스트로

docs/test-list.md에 F2 항목을 추가합니다.

# F2 CsvSalesRepository
- [ ] 정상 CSV를 읽으면 SalesRecord 목록을 반환한다
- [ ] quantity가 음수/0인 행은 제외하고 사유를 기록한다
- [ ] 날짜 형식이 잘못된 행은 제외하고 사유를 기록한다
- [ ] 필수 컬럼이 없으면 명확한 오류를 던진다

 

먼저, 테스트에서 임시 CSV 파일을 손쉽게 만들 픽스처(fixture) 를 둡니다. tests/conftest.py:

from pathlib import Path  # 경로를 객체로 다루는 표준 모듈(문자열 결합보다 안전)

import pytest


@pytest.fixture  # 이 함수를 'pytest 픽스처(준비물 제공자)'로 등록한다
def csv_file(tmp_path: Path):
    """주어진 내용으로 임시 CSV 파일을 만들어 그 '경로'를 돌려주는 팩토리.

    tmp_path: pytest가 테스트마다 자동으로 만들어 주는 '전용 임시 폴더' 픽스처.
              테스트가 끝나면 자동 삭제되므로 실제 프로젝트를 더럽히지 않는다.

    반환값이 '경로'가 아니라 '경로를 만들어 주는 함수(_make)'인 점에 주목.
    덕분에 각 테스트가 원하는 CSV 내용을 그때그때 넣어 파일을 만들 수 있다.
    """

    def _make(content: str) -> Path:
        """content 문자열을 임시 CSV 파일로 쓰고 그 경로를 반환한다."""
        p = tmp_path / "sales.csv"                 # 임시 폴더 안의 파일 경로(/ 연산자로 결합)
        p.write_text(content, encoding="utf-8")    # 한글이 깨지지 않도록 UTF-8로 기록
        return p

    return _make   # 함수 자체를 반환 → 테스트에서 csv_file("...")처럼 호출

📝 픽스처란? pytest에서 "테스트에 필요한 준비물을 만들어 주는 함수"입니다. 테스트 함수의 인자 이름(csv_file)으로 적으면 pytest가 알아서 연결해 줍니다.

 

이제 tests/test_csv_repository.py:

from sales_agent.adapters.csv_repository import CsvSalesRepository


# 모든 테스트 CSV가 공유하는 헤더 줄(컬럼 이름). 끝의 \n은 줄바꿈.
HEADER = "date,product,region,quantity,unit_price\n"


def test_loads_valid_rows(csv_file):
    """정상 행은 SalesRecord로 적재된다."""
    # csv_file(...)에 '헤더 + 정상 데이터 1행' 문자열을 넘겨 임시 CSV를 만든다.
    path = csv_file(HEADER + "2024-01-05,노트북,서울,3,1200000\n")
    repo = CsvSalesRepository(str(path))  # 어댑터에 파일 경로를 주입

    records = repo.load()  # 실제 적재 수행

    assert len(records) == 1               # 정상 행 1건이 적재됨
    assert records[0].product == "노트북"   # 값이 올바르게 매핑됐는가
    assert records[0].revenue == 3_600_000  # 3 × 1,200,000 (도메인 계산이 동작)


def test_skips_invalid_quantity_with_reason(csv_file):
    """수량이 0 이하인 행은 제외되고, 그 사유가 기록된다."""
    path = csv_file(
        HEADER
        + "2024-01-05,노트북,서울,3,1200000\n"   # 정상 행
        + "2024-01-06,마우스,부산,0,25000\n"      # 수량 0 → 제외 대상
    )
    repo = CsvSalesRepository(str(path))

    records = repo.load()

    assert len(records) == 1                 # 정상 행만 남는다(잘못된 행 제외)
    assert len(repo.skipped) == 1            # 제외 보고가 정확히 1건
    assert "quantity" in repo.skipped[0].reason  # 사유에 원인('quantity')이 담겨 있다


def test_skips_bad_date(csv_file):
    """날짜 형식이 잘못된 행은 제외된다."""
    # 2024/13/40 은 존재할 수 없는 날짜(13월 40일) → 변환 실패 → 제외
    path = csv_file(HEADER + "2024/13/40,노트북,서울,3,1200000\n")
    repo = CsvSalesRepository(str(path))

    records = repo.load()

    assert records == []          # 적재된 정상 행이 없다
    assert len(repo.skipped) == 1  # 그 행은 사유와 함께 제외 보고됨


def test_missing_required_column_raises(csv_file):
    """필수 컬럼이 통째로 없으면(데이터 구조 자체가 잘못) 오류를 던진다."""
    import pytest

    # unit_price 컬럼이 아예 없는 CSV → 행 단위 문제가 아니라 '파일 구조' 문제
    path = csv_file("date,product,region,quantity\n2024-01-05,노트북,서울,3\n")
    repo = CsvSalesRepository(str(path))

    with pytest.raises(ValueError):  # 건너뛰지 않고 전체를 중단(예외)해야 정상
        repo.load()

 

🔍 관찰 포인트 — "행 단위 오류"와 "파일 전체 오류"를 구분합니다.

• 한 행만 잘못된 경우(수량 0, 날짜 오류) → 그 행만 건너뛰고 계속 진행합니다(skipped에 기록).
• 필수 컬럼 자체가 없는 경우(unit_price 컬럼 없음) → 파일 구조가 잘못된 것이므로 전체를 오류 처리합니다(ValueError). "어디까지 관용하고 어디서 멈출지"를 테스트로 명확히 정의한 것입니다.

uv run pytest tests/test_csv_repository.py -v   # 🔴 실패 확인

---

🟢 Green — 어댑터 구현

이제 위 테스트를 통과시키는 CsvSalesRepository를 src/sales_agent/adapters/csv_repository.py에 직접 작성합니다. 지켜야 할 규칙은 ▲usecases/ports.py의 SalesRepository 프로토콜을 만족할 것 ▲pandas로 CSV를 읽되 반환은 list[SalesRecord]일 것 ▲잘못된 행은 예외를 잡아 self.skipped에 (행 번호, 사유)로 기록하고 건너뛸 것 ▲필수 컬럼 누락은 ValueError로 던질 것입니다.

 

src/sales_agent/adapters/csv_repository.py:

from __future__ import annotations

from dataclasses import dataclass
from datetime import datetime  # 문자열 → 날짜 변환(strptime)에 사용

import pandas as pd  # 어댑터(3계층)이므로 외부 라이브러리 사용 OK

from sales_agent.domain.errors import InvalidSalesRecordError
from sales_agent.domain.models import SalesRecord

# CSV에 반드시 있어야 하는 컬럼 목록. 하나라도 없으면 '파일 구조 오류'로 본다.
REQUIRED_COLUMNS = ["date", "product", "region", "quantity", "unit_price"]


@dataclass(frozen=True)
class SkippedRow:
    """적재에서 제외된 행 1건의 정보(행 번호 + 사유)를 담는 불변 객체."""

    line: int     # 사용자가 보는 실제 CSV 행 번호(1부터, 헤더 포함)
    reason: str   # 제외된 이유(예: "quantity must be > 0, got 0")


class CsvSalesRepository:
    """CSV 파일에서 매출 레코드를 적재하는 어댑터.

    usecases.ports.SalesRepository 프로토콜(load 메서드)을 구현한다.
    '더러운 CSV'를 받아 '깨끗한 도메인 객체 목록'으로 통역하는 것이 책임이다.
    """

    def __init__(self, path: str) -> None:
        """path: 읽어들일 CSV 파일 경로."""
        self._path = path
        # 적재 중 제외된 행들을 모아 둘 목록. 호출자(화면 등)가 사유를 보여 줄 수 있다.
        self.skipped: list[SkippedRow] = []

    def load(self) -> list[SalesRecord]:
        """CSV를 읽어 유효한 SalesRecord 목록을 반환한다.

        - 한 행만 잘못된 경우 → self.skipped에 사유를 기록하고 그 행만 건너뛴다.
        - 필수 컬럼이 통째로 없는 경우 → ValueError로 전체를 중단한다.
        """
        # dtype=str: 모든 값을 일단 문자열로 읽는다(숫자 자동 변환으로 인한 오작동 방지).
        # fillna(""): 빈 칸(NaN)을 빈 문자열로 바꿔 이후 .strip() 등에서 오류가 안 나게.
        df = pd.read_csv(self._path, dtype=str).fillna("")

        # 필수 컬럼 중 실제 파일에 없는 것들을 모은다.
        missing = [c for c in REQUIRED_COLUMNS if c not in df.columns]
        if missing:  # 하나라도 없으면 파일 구조 자체가 잘못된 것 → 중단
            raise ValueError(f"필수 컬럼 누락: {missing}")

        self.skipped = []                      # 매 호출마다 제외 목록 초기화
        records: list[SalesRecord] = []
        for i, row in df.iterrows():           # i: 0기반 행 인덱스, row: 한 행(Series)
            try:
                records.append(self._to_record(row))  # 변환 성공 → 목록에 추가
            except (InvalidSalesRecordError, ValueError) as e:
                # 변환/검증 실패 → 그 행만 건너뛰고 사유를 기록한다.
                # 실제 행 번호 = 0기반 인덱스(i) + 헤더 1줄 + 1 = i + 2
                self.skipped.append(SkippedRow(line=int(i) + 2, reason=str(e)))
        return records

    @staticmethod
    def _to_record(row: pd.Series) -> SalesRecord:
        """한 행(문자열들)을 적절한 타입으로 변환해 SalesRecord로 만든다.

        여기서 SalesRecord를 생성하는 순간, 도메인의 __post_init__ 검증이 자동 실행된다.
        즉 수량/단가/상품명 규칙 위반은 이 한 줄에서 예외로 걸러진다(검증 중복 없음).
        """
        return SalesRecord(
            # "2024-01-05" 형식 문자열을 date 객체로 변환. 형식이 다르면 여기서 예외.
            date=datetime.strptime(row["date"].strip(), "%Y-%m-%d").date(),
            product=row["product"].strip(),       # 앞뒤 공백 제거
            region=row["region"].strip(),
            quantity=int(row["quantity"]),        # 문자열 → 정수(실패 시 예외)
            unit_price=float(row["unit_price"]),  # 문자열 → 실수(실패 시 예외)
        )

 

🔍 관찰 포인트 1 — 검증 규칙을 도메인에서 재사용합니다. _to_record는 값을 변환한 뒤 SalesRecord(...)를 생성합니다. 이때 5교시에서 만든 도메인의 __post_init__이 음수 수량 등을 자동으로 막아 줍니다. 어댑터는 그 예외를 잡아 사유로 기록할 뿐, 검증 규칙을 다시 작성하지 않습니다. 검증의 단일 원천은 도메인 이라는 원칙을 지킨 것입니다(규칙이 두 곳에 흩어지면 언젠가 어긋납니다).

 

🔍 관찰 포인트 2 — line=int(i)+2의 의미. pandas의 행 인덱스는 0부터 시작하고, CSV의 첫 줄은 헤더입니다. 사용자가 엑셀에서 보는 "실제 행 번호"와 맞추려면 0기반 인덱스 + 헤더 1줄 + 1을 더합니다. 사용자에게 보고할 때 정확한 행을 가리키기 위한 보정입니다.

uv run pytest tests/test_csv_repository.py -v   # 🟢 통과 확인

---

🔵 Refactor + 샘플 데이터 준비

코드를 다듬은 뒤(중복 정리·이름 정돈), 이후 교시에서 계속 쓸 샘플 CSV를 만들어 둡니다.

New-Item -ItemType Directory -Force -Path sample_data

 

VS Code에서 sample_data/sales_sample.csv를 만들고 다음을 입력합니다. (파일을 UTF-8 로 저장하세요.)

date,product,region,quantity,unit_price
2024-01-05,노트북,서울,3,1200000
2024-01-06,마우스,부산,10,25000
2024-02-11,노트북,서울,2,1200000
2024-02-15,키보드,대구,5,89000
2024-03-03,모니터,서울,4,350000
2024-03-21,마우스,부산,0,25000
2024-04-09,노트북,인천,1,1200000

 

🔍 관찰 포인트: 6번째 데이터 행(2024-03-21,마우스,부산,0,25000)은 일부러 넣은 잘못된 데이터(수량 0)입니다. 앱이 이 행을 멈추지 않고 건너뛰며 사유를 보고하는지, 10교시 웹 화면에서 직접 확인하게 됩니다.

uv run pytest -v   # 전체 통과 확인

---

✏️ 미니 실습 — 가짜 리포지토리(Fake)의 가치 미리 보기

7교시의 분석 테스트에서는 CSV 파일조차 필요 없습니다. 포트(SalesRepository)만 만족하면 되기 때문입니다. 아래처럼 "약속만 흉내 내는" 가짜를 만들 수 있습니다.

class FakeSalesRepository:
    """테스트용 가짜 리포지토리. SalesRepository 포트를 만족한다.

    진짜 CsvSalesRepository와 달리 파일을 전혀 읽지 않고,
    생성 시 받은 레코드 목록을 그대로 돌려준다. load() 메서드 모양만 같으면
    유스케이스는 '진짜'인지 '가짜'인지 구분하지 못한다(= 포트의 힘).
    """

    def __init__(self, records):
        self._records = records   # 미리 준비한 SalesRecord 목록을 보관

    def load(self):
        return self._records      # 파일 입출력 없이 즉시 반환

 

🔍 관찰 포인트: 이 Fake를 유스케이스에 끼우면, 파일을 만들지 않고도 분석 로직만 따로 검증할 수 있습니다. 3교시에서 배운 포트/어댑터 분리가 주는 직접적인 이득입니다 — "테스트하기 쉬움"이 설계에서 자연스럽게 따라옵니다.

🔒 진행 게이트

• 제출물(기록): Fake 리포지토리로 파일 없이 레코드를 반환받은 출력(예: 1 3600000).
• 이 결과가 있어야 다음 교시로 넘어갑니다.
• 정답·해설(별도 파일): solutions/미니실습정답_06.md

---

✅ 체크포인트

• [ ] 정상 CSV가 SalesRecord 목록으로 적재된다
• [ ] 잘못된 행이 제외되고 repo.skipped에 사유가 남는다
• [ ] 필수 컬럼 누락 시 ValueError가 난다
• [ ] 어댑터가 pandas를 쓰되 반환은 도메인 객체(SalesRecord)다
• [ ] sample_data/sales_sample.csv가 준비됐다(UTF-8)

---

🛠️ 트러블슈팅

증상	원인	해결
UnicodeDecodeError	CSV 인코딩 불일치	encoding="utf-8" 또는 utf-8-sig 시도
한글 깨짐(엑셀 저장본)	엑셀은 종종 cp949로 저장	pd.read_csv(path, encoding="cp949")로 분기
모든 행이 skipped	날짜 포맷 불일치	CSV의 date가 YYYY-MM-DD인지 확인
int() 변환 실패	quantity에 소수점/문자 혼입	사유로 제외되는 것이 정상. 데이터 점검

---

🔑 핵심 정리

• 어댑터는 지저분한 바깥을 깨끗한 도메인으로 통역하는 차단막이다.
• 잘못된 데이터는 버리지 않고 제외 사유와 함께 보고한다(행 단위 관용, 구조 오류는 중단).
• 도메인 검증을 재사용하므로 검증 규칙이 한 곳(단일 원천) 에 모인다.

다음 교시: 적재된 레코드로 핵심 분석 지표를 계산하는 유스케이스를 TDD로 만듭니다(Fake 리포지토리 활용).