콘텐츠 해시 파일 캐시 패턴
비용이 많이 드는 파일 처리 결과(PDF 파싱, 텍스트 추출, 이미지 분석)를 SHA-256 콘텐츠 해시를 캐시 키로 사용하여 캐싱합니다. 경로 기반 캐싱과 달리 파일 이동/이름 변경에도 캐시가 유지되며 콘텐츠가 변경되면 자동으로 무효화됩니다.
활성화 시기
- 파일 처리 파이프라인 구축(PDF, 이미지, 텍스트 추출)
- 처리 비용이 높고 동일한 파일이 반복적으로 처리될 때
--cache/--no-cacheCLI 옵션이 필요할 때- 기존 순수 함수를 수정하지 않고 캐싱을 추가하려 할 때
핵심 패턴
1. 콘텐츠 해시 기반 캐시 키
파일 경로가 아닌 파일 콘텐츠를 캐시 키로 사용합니다:
import hashlib
from pathlib import Path
_HASH_CHUNK_SIZE = 65536 # 대용량 파일 처리를 위한 64KB 청크
def compute_file_hash(path: Path) -> str:
"""파일 콘텐츠의 SHA-256 해시 (대용량 파일은 청크 단위 처리)."""
if not path.is_file():
raise FileNotFoundError(f"파일을 찾을 수 없음: {path}")
sha256 = hashlib.sha256()
with open(path, "rb") as f:
while True:
chunk = f.read(_HASH_CHUNK_SIZE)
if not chunk:
break
sha256.update(chunk)
return sha256.hexdigest()
왜 콘텐츠 해시인가? 파일 이름 변경/이동 = 캐시 적중. 콘텐츠 변경 = 자동 무효화. 인덱스 파일 불필요.
2. 불변 데이터클래스로 캐시 항목 정의
from dataclasses import dataclass
@dataclass(frozen=True, slots=True)
class CacheEntry:
file_hash: str
source_path: str
document: ExtractedDocument # 캐시된 결과
3. 파일 기반 캐시 저장소
각 캐시 항목은 {hash}.json으로 저장됩니다 — 해시로 O(1) 조회, 인덱스 파일 불필요.
import json
from typing import Any
def write_cache(cache_dir: Path, entry: CacheEntry) -> None:
cache_dir.mkdir(parents=True, exist_ok=True)
cache_file = cache_dir / f"{entry.file_hash}.json"
data = serialize_entry(entry)
cache_file.write_text(json.dumps(data, ensure_ascii=False), encoding="utf-8")
def read_cache(cache_dir: Path, file_hash: str) -> CacheEntry | None:
cache_file = cache_dir / f"{file_hash}.json"
if not cache_file.is_file():
return None
try:
raw = cache_file.read_text(encoding="utf-8")
data = json.loads(raw)
return deserialize_entry(data)
except (json.JSONDecodeError, ValueError, KeyError):
return None # 손상된 경우 캐시 미스로 처리
4. 서비스 계층 래퍼 (SRP)
처리 함수는 순수하게 유지하고 캐싱은 별도의 서비스 계층으로 추가합니다.
def extract_with_cache(
file_path: Path,
*,
cache_enabled: bool = True,
cache_dir: Path = Path(".cache"),
) -> ExtractedDocument:
"""서비스 계층: 캐시 확인 -> 추출 -> 캐시 쓰기."""
if not cache_enabled:
return extract_text(file_path) # 순수 함수, 캐시 인식 불필요
file_hash = compute_file_hash(file_path)
# 캐시 확인
cached = read_cache(cache_dir, file_hash)
if cached is not None:
logger.info("캐시 적중: %s (hash=%s)", file_path.name, file_hash[:12])
return cached.document
# 캐시 미스 -> 추출 -> 저장
logger.info("캐시 미스: %s (hash=%s)", file_path.name, file_hash[:12])
doc = extract_text(file_path)
entry = CacheEntry(file_hash=file_hash, source_path=str(file_path), document=doc)
write_cache(cache_dir, entry)
return doc
주요 설계 결정
| 결정 | 이유 |
|---|---|
| SHA-256 콘텐츠 해시 | 경로 독립적, 콘텐츠 변경 시 자동 무효화 |
{hash}.json 파일 명명 | O(1) 조회, 인덱스 파일 불필요 |
| 서비스 계층 래퍼 | SRP: 추출은 순수, 캐시는 별도 책임 |
| 수동 JSON 직렬화 | 불변 데이터클래스 직렬화 완전 제어 |
손상 시 None 반환 | 우아한 저하, 다음 실행 시 재처리 |
cache_dir.mkdir(parents=True) | 첫 쓰기 시 지연 디렉토리 생성 |
모범 사례
- 콘텐츠를 해싱하고 경로는 해싱하지 마세요 — 경로는 변경되지만 콘텐츠 정체성은 변하지 않습니다
- 대용량 파일은 청크 단위로 해싱 — 전체 파일을 메모리에 로드하지 마세요
- 처리 함수는 순수하게 유지 — 캐싱에 대해 알 필요가 없습니다
- 캐시 적중/미스를 로깅 — 디버깅을 위해 해시의 앞부분만 사용
- 손상을 우아하게 처리 — 잘못된 캐시 항목은 미스로 처리하고 절대 중단하지 마세요
피해야 할 안티 패턴
# 나쁨: 경로 기반 캐싱 (파일 이동/이름 변경 시 깨짐)
cache = {"/path/to/file.pdf": result}
# 나쁨: 처리 함수 내부에 캐시 로직 추가 (SRP 위반)
def extract_text(path, *, cache_enabled=False, cache_dir=None):
if cache_enabled: # 이제 이 함수는 두 가지 책임을 가짐
...
# 나쁨: 중첩된 불변 데이터클래스와 함께 dataclasses.asdict() 사용
# (복잡한 중첩 타입에서 문제 발생 가능)
data = dataclasses.asdict(entry) # 대신 수동 직렬화 사용
사용 시기
- 파일 처리 파이프라인 (PDF 파싱, OCR, 텍스트 추출, 이미지 분석)
--cache/--no-cache옵션이 유용한 CLI 도구- 동일한 파일이 여러 실행에 걸쳐 나타나는 배치 처리
- 기존 순수 함수를 수정하지 않고 캐싱 추가
사용하지 말아야 할 시기
- 항상 최신 데이터가 필요한 경우 (실시간 피드)
- 캐시 항목이 매우 클 경우 (스트리밍 고려)
- 파일 콘텐츠 외의 매개변수에 따라 결과가 달라지는 경우 (예: 다른 추출 설정)