콘텐츠 해시 파일 캐시 패턴
비용이 많이 드는 파일 처리 결과(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 도구- 여러 실행에 걸쳐 동일한 파일이 나타나는 배치 처리
- 기존 순수 함수를 수정하지 않고 캐싱 추가
사용하지 말아야 할 경우
- 항상 최신 데이터가 필요한 경우 (실시간 피드)
- 캐시 항목이 매우 큰 경우 (대신 스트리밍 고려)
- 파일 콘텐츠 외부의 매개변수에 의존하는 결과 (예: 다른 추출 설정)