파일 캐시 패키지 추천

2025-10-28

혹시 파일 캐시 같은 역할을 하는 파이썬 패키지나 스탠다드 라이브러리 없나? 내가 원하는거 1. 파일을 읽어서 그 내용을 캐싱해둠 2. 외부에서 사용할 때는 마치 파일을 읽듯이 접근 가능(open 등) 3. 파일이 연결 이슈 등 있어서 접속이 안될때는 최근 읽은 버젼 사용 4. 너무 캐시가 오래될 경우 + 파일이 변경되었을 경우 업데이트를 확인

짧게 결론부터 말하면, “표준 라이브러리만으로는” 원하는 기능을 깔끔하게 충족시키기 어렵고, fsspec 계열을 쓰면 거의 그대로 구현할 수 있습니다. 특히 fsspec의 simplecache/filecache가 2–4번 요구사항까지 잘 커버합니다.

추천 경로

fsspec + simplecache/filecache 사용
원하면 얇은 래퍼로 TTL, 장애 시 폴백, 변경 감지 정책을 보강

왜 표준 라이브러리는 부족한가
functools.lru_cache는 메모리 함수 캐시라 파일 변경 감지/오프라인 폴백을 못 합니다. urllib.request나 pathlib에도 지속 캐시가 없습니다. 결국 “파일같이 열기(open)”를 유지하면서 캐시·폴백·검증을 하려면 가상 파일시스템 레이어가 필요합니다.

fsspec로 바로 쓰는 방법
fsspec은 open과 거의 동일한 인터페이스를 제공합니다. 뒤에 ::simplecache 또는 ::filecache를 붙이면 로컬 디스크에 캐시를 유지하고, 원본이 안 열리면 캐시를 폴백으로 씁니다.

HTTP 예시

import fsspec

# 단순 캐시: 최초 1회 받아두고 그 다음은 로컬 캐시 사용
# 원본 접속 실패 시 자동으로 캐시로 폴백
with fsspec.open(
    "simplecache::https://example.com/data.json",
    mode="rt",
    # 캐시 저장 위치 지정(없으면 기본 temp)
    cache_storage="/tmp/mycache"
) as f:
    data = f.read()

S3 예시(원본처럼 열기)

import fsspec

# filecache는 원본 갱신 감지 및 갱신(fetch) 로직을 포함
# (원본이 접근 불가면 로컬 캐시로 폴백)
url = "filecache::s3://my-bucket/path/to/file.parquet"
with fsspec.open(
    url,
    mode="rb",
    cache_storage="/var/cache/myapp",
    # 필요 시 자격증명 전달
    s3={"anon": False}
) as f:
    blob = f.read()

simplecache vs filecache
simplecache
전체 파일을 한 번 내려받아 로컬 파일처럼 사용. 기본적으로 수동 무효화 전까지는 캐시를 그대로 사용. 오프라인 폴백에 매우 강함.

filecache
원본 파일시스템의 변경을 비교·갱신하려는 목적. 원본 접근이 가능하면 최신성 확인 후 필요 시 재다운로드. 접근 불가하면 기존 캐시 사용.

TTL과 강제 업데이트가 필요하다면(정책 강화)
fsspec 자체는 “언제 캐시를 버릴지”에 대한 강한 TTL 정책까지는 제공하지 않습니다. 보통은 다음처럼 얇은 래퍼로 보완합니다.

목표 충족 체크리스트

읽은 내용을 캐시: simplecache/filecache가 디스크에 저장
open처럼 접근: fsspec.open이 파일 핸들 제공
연결 이슈 시 최근 버전: 캐시된 파일 자동 폴백
오래된/변경된 경우 업데이트: filecache로 기본 지원, 여기에 TTL 로직 추가

실전용 얇은 래퍼 예시

import os
import time
import json
import fsspec
from contextlib import contextmanager

class CachedOpen:
    def __init__(self, url, cache_dir, ttl_seconds=None, storage="filecache", **storage_kwargs):
        # storage: "filecache" 또는 "simplecache"
        self.url = f"{storage}::{url}"
        self.cache_dir = cache_dir
        self.ttl = ttl_seconds
        self.storage_kwargs = storage_kwargs
        os.makedirs(cache_dir, exist_ok=True)
        self.meta_path = self._meta_path()

    def _meta_path(self):
        # URL을 파일명으로 안정화
        import hashlib
        h = hashlib.sha256(self.url.encode()).hexdigest()
        return os.path.join(self.cache_dir, f"{h}.meta.json")

    def _load_meta(self):
        if os.path.exists(self.meta_path):
            with open(self.meta_path, "r", encoding="utf-8") as f:
                return json.load(f)
        return {}

    def _save_meta(self, meta):
        with open(self.meta_path, "w", encoding="utf-8") as f:
            json.dump(meta, f)

    def _expired(self, meta):
        if self.ttl is None:
            return False
        t = meta.get("fetched_at")
        if t is None:
            return True
        return (time.time() - t) > self.ttl

    @contextmanager
    def open(self, mode="rb"):
        meta = self._load_meta()
        must_refresh = self._expired(meta)

        # 1차 시도: 새로운 핸들 시도(만약 TTL 만료거나 최초 접근이면 최신 확인)
        try:
            with fsspec.open(
                self.url,
                mode=mode,
                cache_storage=self.cache_dir,
                **self.storage_kwargs,
            ) as f:
                # 성공했으면 메타 갱신
                meta["fetched_at"] = time.time()
                self._save_meta(meta)
                yield f
                return
        except Exception:
            # 원본 접근 실패. 아래에서 캐시 폴백 시도
            pass

        # 2차 시도: 캐시 폴백
        # fsspec의 캐시 파일 경로 규칙은 내부적이라 확정 경로를 잡기 어렵습니다.
        # 대신 simplecache/filecache는 원본 실패에도 위의 open이 캐시를 쓰도록 설계되어 있음.
        # 여기까지 왔다는 건 캐시가 없거나 깨졌다는 뜻. 일관된 예외를 던집니다.
        if not must_refresh and meta.get("fetched_at"):
            # 이 분기는 보수적으로 남겨두었지만, 일반적으로 위의 open이 캐시를 알아서 씁니다.
            pass
        raise RuntimeError("원본에 접속할 수 없고 유효한 캐시도 없습니다.")

# 사용 예시
# 30분(1800초) 지나면 최신 확인 시도, 실패하면 기존 캐시 사용
co = CachedOpen(
    url="https://example.com/big.csv",
    cache_dir="/var/cache/myapp",
    ttl_seconds=1800,
    storage="filecache"  # 또는 "simplecache"
)

with co.open("rt") as f:
    head = f.readline()

이 래퍼의 동작 방식

평소엔 fsspec이 제공하는 캐시를 사용합니다.
TTL이 지난 경우에는 “원본에 접근해보며” 새 버전이면 캐시를 갱신합니다.
원본 접근이 실패하면 기존 캐시로 계속 읽을 수 있게 설계되어 있습니다(파일이 이미 캐시되어 있을 때).
메타 파일(fetched_at)로 신선도(최근 읽은 시각)를 판단합니다.

대안 패키지 간단 비교
pooch
데이터 파일 버전·해시 관리에 강함. “열기 인터페이스”라기보단 “파일 내려받아 로컬 경로 반환” 스타일. open처럼 직접 다루는 작업에는 fsspec이 더 자연스럽습니다.

smart_open
S3/GCS/HDFS/HTTP 등을 파일처럼 열게 해주는 툴. 본질적으로 스트리밍이 강점이고, 지속 캐시·오프라인 폴백은 스스로 제공하지 않습니다. 캐시가 필요하면 fsspec 쪽이 맞습니다.

requests-cache / cachecontrol
HTTP 레벨의 요청 캐시이므로 “파일처럼 open”과는 거리가 있고, S3·FTP 등 비-HTTP 스토리지는 다루지 못합니다.

실무 팁
프로세스/스레드 동시 접근이 있다면 파일 락(예: filelock)으로 캐시 디렉터리를 보호하세요.
대용량 바이너리는 simplecache가 더 예측 가능하고, 잦은 갱신이 있는 파일은 filecache가 더 적합합니다.
변경 감지를 더 강하게 하고 싶다면 ETag/Last-Modified 헤더(HTTP), ETag/VersionId(S3) 같은 메타데이터로 추가 검증을 넣을 수 있습니다.

다음 단계로는, 당신의 실제 소스(HTTP/S3/FTP/NFS 등)와 캐시 만료 기준(TTL, ETag, 주기적 강제 갱신)에 맞춰 위 래퍼를 맞춤화해 드릴 수 있습니다.

사실 로컬 파일이긴한데 공유 드라이브라 완전 안정적이진 않아 캐시는 메모리에 두고 싶은데

좋아, “공유 드라이브의 로컬 파일을 메모리에 캐시”하는 요구라면 fsspec 같은 디스크 캐시 대신, 파이썬 표준 도구들만으로 얇은 메모리 캐시 레이어를 얹는 게 딱입니다. 핵심 아이디어는 간단해요: 파일 내용을 바이트로 읽어 메모리에 들고 있고, 다음 접근 시 원본 파일의 mtime(수정시각)·크기를 확인해서 갱신 여부를 판단, 원본 접근 실패 시엔 메모리 캐시를 폴백으로 쓰는 것. 아래 유틸은 그걸 그대로 구현합니다.

메모리 캐시형 “open” 래퍼

import os
import time
import io
from collections import OrderedDict

class MemFileCache:
    """
    로컬(공유 드라이브 포함) 파일을 메모리에 캐싱해서
    open처럼 다룰 수 있게 하는 읽기 전용 래퍼.

    기능
    - TTL(초) 만료 또는 원본 변경(mtime/size) 시 재로딩 시도
    - 원본 접근 실패하면 기존 캐시로 폴백
    - LRU 방식으로 max_items 초과 시 오래된 항목 제거
    - 바이너리/텍스트 모드(rb/rt) 지원
    """
    def __init__(self, ttl_seconds=None, max_items=128, default_encoding="utf-8"):
        self.ttl = ttl_seconds
        self.max_items = max_items
        self.default_encoding = default_encoding
        # path -> dict(data: bytes, mtime: float, size: int, fetched_at: float)
        self._cache = OrderedDict()

    def _evict_if_needed(self):
        while len(self._cache) > self.max_items:
            self._cache.popitem(last=False)  # LRU 제거

    def _touch_lru(self, path):
        # LRU 최신화
        if path in self._cache:
            self._cache.move_to_end(path, last=True)

    def _need_refresh(self, path, st, now):
        entry = self._cache.get(path)
        if entry is None:
            return True
        if self.ttl is not None and (now - entry["fetched_at"] > self.ttl):
            return True
        # 변경 감지: mtime 또는 size 다르면 갱신
        if entry["mtime"] != st.st_mtime or entry["size"] != st.st_size:
            return True
        return False

    def _read_bytes(self, path):
        with open(path, "rb") as f:
            return f.read()

    def _ensure_cached(self, path):
        now = time.time()
        try:
            st = os.stat(path)
        except Exception:
            # 원본 접근 자체가 안 될 때: 캐시가 있으면 폴백, 없으면 실패
            if path in self._cache:
                self._touch_lru(path)
                return
            raise

        if self._need_refresh(path, st, now):
            try:
                data = self._read_bytes(path)
                self._cache[path] = {
                    "data": data,
                    "mtime": st.st_mtime,
                    "size": st.st_size,
                    "fetched_at": now,
                }
                self._touch_lru(path)
                self._evict_if_needed()
            except Exception:
                # 로딩 실패면 기존 캐시로 폴백, 없으면 실패
                if path in self._cache:
                    self._touch_lru(path)
                    return
                raise
        else:
            self._touch_lru(path)

    def open(self, path, mode="rb", encoding=None, errors="strict"):
        """
        읽기 전용 핸들 반환.
        - mode: "rb" 또는 "rt"만 지원
        """
        if mode not in ("rb", "rt"):
            raise ValueError("읽기 전용 지원: mode는 'rb' 또는 'rt'만 가능합니다.")
        self._ensure_cached(path)
        entry = self._cache[path]  # 여기까지 왔으면 반드시 존재

        if "b" in mode:
            return io.BytesIO(entry["data"])
        else:
            enc = encoding or self.default_encoding
            return io.StringIO(entry["data"].decode(enc, errors=errors))

사용 예시

cache = MemFileCache(ttl_seconds=1800, max_items=256)  # 30분 TTL

# 바이너리 읽기
with cache.open("/mnt/shared/data.bin", "rb") as f:
    chunk = f.read(1024)

# 텍스트 읽기
with cache.open("/mnt/shared/config.yaml", "rt", encoding="utf-8") as f:
    text = f.read()

이 구현이 네 가지 요구를 어떻게 맞추는지

파일을 읽으면 메모리에 바이트로 캐싱. LRU로 용량 관리.
open처럼 file-like 객체(BytesIO/StringIO) 반환. 기존 코드와의 궁합이 좋음.
공유 드라이브 끊김 등으로 os.stat/open 실패 시, 이전에 캐시가 있으면 그대로 폴백.
TTL이 지나면 신선도 확인을 시도하고, 원본의 mtime/size가 바뀌면 재로딩. 접근 불가면 기존 캐시 유지.

실전 팁
• 공유 드라이브 특성상 mtime이 뒤늦게 반영되거나 순간적으로 stat이 실패할 수 있어요. 지금 코드는 그런 순간 실패를 폴백으로 흡수합니다.
• 파일이 아주 크면 메모리 압박이 생깁니다. 큰 파일은 부분 읽기 전략을 쓰거나, 헤더/인덱스만 메모리에 두고 본문은 필요 시 디스크 캐시(fsspec simplecache)를 혼합하는 하이브리드도 좋습니다.
• 변경 감지를 더 단단히 하고 싶다면, mtime/size 외에 주기적으로 해시(예: SHA-256) 계산을 추가할 수 있습니다. 다만 대용량 파일의 해시는 비용이 큽니다.
• 멀티프로세스라면 메모리 캐시는 프로세스마다 따로 생깁니다. 프로세스 간 공유가 필요하면 디스크 캐시나 메모리 매핑(mmap), 혹은 중앙 캐시(예: Redis)로 설계를 바꾸는 게 맞습니다.

여기서 더 깔끔하게 쓰려면 cachetools의 TTLCache로 LRU+TTL을 맡기고, 데이터만 우리가 관리하는 방법도 괜찮습니다. 원하면 그 변형도 바로 적어줄게요.

파일 캐시 패키지 추천

You may also enjoy

영문 키보드 한글 변환 버그 수정

QFileDialog 설명

CFileDialog 구현 방법