Pythonでキャッシュファイルをストレージに書き出して出力を再利用する


Pythonの標準ライブラリには、`cache`や`lru_cache`など、メモリにキャッシュを書き出す関数はありますが、ストレージに書き出す関数はありません。
以下は`cache`の利用例です

```python
from time import sleep
from functools import cache
from datetime import datetime
@cache
def any_func()->datetime:
    return datetime.now()
print(any_func())
sleep(1)
print(any_func())
```

```bash:出力
2024-04-13 16:19:55.398819
2024-04-13 16:19:55.398819
```

頻繁にストレージに保存するタイプのキャッシュ処理を書くことがあり、毎回新規で書くのも面倒なのでメモ書きとして書き残します。

```python
from pathlib import Path
import pickle
from typing import Callable, TypeVar, TypeVarTuple, Unpack, Annotated, Any
import hashlib
import inspect
from functools import wraps

R = TypeVar("R")
Ts = TypeVarTuple("Ts")

class StorageCache:
    def __init__(self, save_dir: Path = Path(".cache")) -> None:
        self.save_dir = save_dir
        self.save_dir.mkdir(parents=True, exist_ok=True)

    def _sort_kwargs(self, func, *args, **kwargs) -> tuple[list[Any], dict[str, Any]]:
        """キーワード引数を関数の引数の順序に合わせてソートします。

        Parameters
        ----------
        func : Callable
            引数をソートする関数

        Returns
        -------
        tuple[list[Any], dict[str, Any]]
            ソートされた引数とキーワード引数を返します
        """
        argspec = inspect.getfullargspec(func)
        for arg in argspec.args:
            if arg in kwargs:
                args = [kwargs[arg], *args]
                del kwargs[arg]
        return args, kwargs

    def get_cache_file(
        self, func: Callable[[Unpack[Ts]], R], *args: Unpack[Ts], **kwargs: dict[str, Any]
    ) -> Path:
        """関数と引数のキャッシュファイルを取得します。

        Parameters
        ----------
        func : Callable[[Unpack[Ts]], R]
            キャッシュファイルを取得する関数

        Returns
        -------
        Path
            キャッシュファイルを返します
        """
        # 引数のハッシュを作成
        args, kwargs = self._sort_kwargs(func, *args, **kwargs)
        arg_hash = hashlib.md5(
            "".join(
                [str(hash(arg)) for arg in args]
                + [str(hash(v)) for v in kwargs.values()]
            ).encode("utf-8")
        ).hexdigest()
        # モジュール名のハッシュを作成
        mod_hash = hashlib.md5(func.__module__.encode()).hexdigest()
        cache_file = self.save_dir / f"{func.__name__}_{mod_hash}/{arg_hash}.pkl"
        return cache_file

    def __call__(self, func: Callable[[Unpack[Ts]], R]) -> Callable[[Unpack[Ts]], R]:
        """関数の結果をキャッシュするデコレーター。"""
        @wraps(func)
        def cached_func(*args, **kwargs) -> R:
            cache_file = self.get_cache_file(func, *args, **kwargs)
            # キャッシュファイルが存在する場合、キャッシュされた結果を返す
            if cache_file.exists():
                return pickle.loads(cache_file.read_bytes())
            # そうでなければ、関数を呼び出して結果を保存
            cache_file.parent.mkdir(parents=True, exist_ok=True)
            result = func(*args, **kwargs)
            cache_file.write_bytes(pickle.dumps(result))
            return result

        return cached_func
```

保存するキャッシュはパラメータの与え方によらず一定のフォルダに保存するようにしています。
引数のソート処理も書いてあるので、引数の与える順番が違っても同じ結果が返ってくるようになっています。

```python
from ucimlrepo import fetch_ucirepo

print(StorageCache().get_cache_file(fetch_ucirepo, 9, "auto+mpg"))
print(StorageCache().get_cache_file(fetch_ucirepo, name="auto+mpg", id=9))
print(StorageCache().get_cache_file(fetch_ucirepo, id=9, name="auto+mpg"))
```

```:出力
.cache\fetch_ucirepo_62c1c82607c8cfcdc8877739eafa0faf\8051318f03407f898d86134f875a6fc7.pkl
.cache\fetch_ucirepo_62c1c82607c8cfcdc8877739eafa0faf\8051318f03407f898d86134f875a6fc7.pkl
.cache\fetch_ucirepo_62c1c82607c8cfcdc8877739eafa0faf\8051318f03407f898d86134f875a6fc7.pkl
```

もし、使う機会があったらいいねください（乞食）