【Python】関数の実行結果をプロセスをまたいでキャッシュする方法

  • 2023年3月21日
  • 2023年3月21日
  • Python

 Python でプログラミングをしている際、しばしばある値をある値に変換する関数を作る時があります。例えば次の様なものです。

def repeat(text: str) -> str:
    return f'{text}{text}'

 実際はもっと複雑ですが、要するに入力と出力のペアが一意に定まり副作用を持たない関数です。こういった関数は一度ある引数で関数を実行すると二回目以降は関数の中の計算処理をスキップしたくなります。これを実現する方法の実装例を紹介します。ちなみに紹介する方法は昨今、流行っているAIなどの機械学習でもよく使われるタイプのキャッシュ機構です。学習などではキャッシュで使う領域や目的からキャッシュの寿命や読み書き方法が割と変わりますが大枠は大体こんな感じです。

 実装の方針としては関数のある場所と名前と引数をキーにして返り値をキャッシュ化し、もしキャッシュが存在するならば関数内の処理を実行しない、というものです。これを実際に実装した例が次です。

import hashlib
import os
import pickle
from typing import Any, Callable


def sha256hasher(text: str) -> str:
    """
    文字列を sha256 ハッシュにする。
    Windows上でファイルとして保存可能な文字列に変換する目的で用意

    Args:
        text (str): ハッシュ化する文字列

    Returns:
        str: ハッシュ値
    """
    text_bytes = text.encode('utf-8')
    hash_object = hashlib.sha256(text_bytes)
    return hash_object.hexdigest()


# キャッシュファイルを置くディレクトリのパス
CACHE_DIR = f'{os.path.dirname(os.path.abspath(__file__))}/cache'


def cached(func: Callable) -> Callable:
    """
    引数に与えられた関数をキャッシュするデコレータ。

    Args:
        func (Callable): キャッシュしたい関数

    Returns:
        Callable: 引数で与えられた関数のキャッシュを行う関数
    """

    def wrapper(*args: Any, **kwargs: Any) -> Any:
        """
        キャッシュをするためのラッパー関数。

        Args:
            *args: 引数
            **kwargs: キーワード引数

        Returns:
            Any: 引数で与えられた関数の戻り値

        """
        # 関数コード、引数、キーワード引数を連結してSHA256ハッシュ値を算出する
        # このハッシュ値がキャッシュファイルの名前であり、キーであるようにした
        # 関数の中身、引数やキーワード引数が変わった場合でも別のハッシュ値になるようにした
        # 関数名やモジュール名を入れないことで完全なコピペ関数は同一のキャッシュファイルを読むようにした
        func_hash = sha256hasher('-'.join([
            str(func.__code__.co_code),  # 関数のコンパイルされたバイトコード
            str(args),  # 位置引数の値
            str(kwargs),  # キーワード引数の値
        ]))
        # キャッシュのパスを生成
        # 単一のディレクトリにガンガンファイルが増えるのが問題になるなら func_hash をn文字毎に区切ってディレクトリを分ける様にすると良いやも
        cache_path = os.path.join(CACHE_DIR, f'{func_hash}.pkl')
        if os.path.exists(cache_path):
            # キャッシュが存在する場合は、デシリアライズして返す
            with open(cache_path, 'rb') as f:
                result = pickle.load(f)
        else:
            # キャッシュが存在しない場合は、関数を実行してシリアライズして保存する
            result = func(*args, **kwargs)
            with open(cache_path, 'wb') as f:
                pickle.dump(result, f)
        return result

    return wrapper

### 使用例 ###

# デコレーターでキャッシュ機構を使うと宣言
@cached
def a(text):
    return text


@cached
def b(text):
    return text


if __name__ == '__main__':
    print(a('a'))
    print(a('a')) # キャッシュから読み込まれる
    print(b('a')) # 中身が同じ関数かつ同じ引数なので、キャッシュから読み込まれる

 Python の関数の持つオブジェクトとしての側面を利用してキャッシュを作ります。こうすると同じ働きをする関数を同じように呼び出すとあっという間に結果を得られ、計算資源も節約できます。この様なキャッシュを作り、読み込む仕組みを作っておくとAPIやデータベース、長大な計算を介するプログラムの利用が大分快適になります。

>株式会社シーポイントラボ

株式会社シーポイントラボ

TEL:053-543-9889
営業時間:9:00~18:00(月〜金)
住所:〒432-8003
   静岡県浜松市中央区和地山3-1-7
   浜松イノベーションキューブ 315
※ご来社の際はインターホンで「316」をお呼びください

CTR IMG