Pythonの標準ライブラリには、cacheやlru_cacheなど、メモリにキャッシュを書き出す関数はありますが、ストレージに書き出す関数はありません。
以下はcacheの利用例です
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())
出力
2024-04-13 16:19:55.398819
2024-04-13 16:19:55.398819
頻繁にストレージに保存するタイプのキャッシュ処理を書くことがあり、毎回新規で書くのも面倒なのでメモ書きとして書き残します。
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
保存するキャッシュはパラメータの与え方によらず一定のフォルダに保存するようにしています。
引数のソート処理も書いてあるので、引数の与える順番が違っても同じ結果が返ってくるようになっています。
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
もし、使う機会があったらいいねください（乞食）