Python Caching Mechanism and functools lru_cache

Caching is a processing method that stores a certain amount of data in preparation for subsequent retrieval needs, aiming to speed up data retrieval. The process of generating data may involve operations such as computation, organization, and remote fetching. If the same piece of data needs to be used multiple times, regenerating it each time would greatly waste time. Therefore, if data obtained from operations such as computation or remote requests is cached, subsequent data retrieval needs will be sped up.

Let's start with a simple example to understand the concept of a caching mechanism:

# -*- coding: utf-8 -*-

import random
import datetime

class MyCache:
    """缓存类"""

    def __init__(self):
        # 用字典结构以 kv 的形式缓存数据
        self.cache = {}
        # 限制缓存的大小，因为缓存的空间有限
        # 所以当缓存太大时，需要将旧的缓存舍弃掉
        self.max_cache_size = 10

    def __contains__(self, key):
        """根据该键是否存在于缓存当中返回 True 或者 False"""
        return key in self.cache

    def get(self, key):
        """从缓存中获取数据"""
        data = self.cache[key]
        data["date_accessed"] = datetime.datetime.now()
        return data["value"]

    def add(self, key, value):
        """更新该缓存字典，如果缓存太大则先删除最早条目"""
        if key not in self.cache and len(self.cache) >= self.max_cache_size:
            self.remove_oldest()
        self.cache[key] = {
            'date_accessed': datetime.datetime.now(),
            'value': value
        }

    def remove_oldest(self):
        """删除具备最早访问日期的输入数据"""
        oldest_entry = None

        for key in self.cache:
            if oldest_entry is None:
                oldest_entry = key
                continue
            curr_entry_date = self.cache[key]['date_accessed']
            oldest_entry_date = self.cache[oldest_entry]['date_accessed']
            if curr_entry_date < oldest_entry_date:
                oldest_entry = key

        self.cache.pop(oldest_entry)

    @property
    def size(self):
        """返回缓存容量大小"""
        return len(self.cache)

if __name__ == '__main__':
    # 测试缓存功能
    cache = MyCache()
    cache.add("test", sum(range(100000)))
    assert cache.get("test") == cache.get("test")

    keys = [
        'red', 'fox', 'fence', 'junk', 'other', 'alpha', 'bravo', 'cal',
        'devo', 'ele'
    ]
    s = 'abcdefghijklmnop'
    for i, key in enumerate(keys):
        if key in cache:
            continue
        else:
            value = ''.join([random.choice(s) for i in range(20)])
            cache.add(key, value)

    assert "test" not in cache
    print(cache.cache)

The example above simply demonstrates the principle of a caching mechanism: data is placed in a dictionary as key-value pairs, so that the next time a value is needed it can be retrieved directly from the dictionary. The implementation of deleting old data in this example is not efficient; in real applications, other approaches can be used.

In Python 3.2, a very elegant caching mechanism was introduced, namely the lru_cache decorator in the functool module, which can directly cache the results of functions or class methods, and subsequent calls then return the cached results directly. The prototype of lru_cache is as follows:

@functools.lru_cache(maxsize=None, typed=False)

Using the lur_cache decorator from the functools module, you can cache up to maxsize call results of this function, thereby improving program execution efficiency, which is especially suitable for time-consuming functions. The parameter maxsize is the maximum number of cached calls; if it is None, there is no limit, and performance is best when set to a power of 2; if typed=True (note that this parameter does not exist in functools32), calls with different argument types will be cached separately, for example f(3) and f(3.0).

LRU (Least Recently Used) is a cache eviction strategy. It performs eviction based on the historical access records of the data; the core idea is, “if data has been accessed recently, then the probability of it being accessed in the future is also higher.” This algorithm was originally a page replacement algorithm for memory management in operating systems, mainly used to find memory blocks in memory that have not been used for a relatively long time and move them out of memory to provide space for new data. Its principle is like the simple example above.

Functions decorated with lru_cache will have two methods, cache_clear and cache_info, used respectively to clear the cache and view cache information. The following is a simple demonstration of using lru_cache:

from functools import lru_cache

@lru_cache(None)
def add(x, y):
    print("calculating: %s + %s" % (x, y))
    return x + y

print(add(1, 2))
print(add(1, 2))
print(add(2, 3))

Output:

calculating: 1 + 2
3
3
calculating: 2 + 3
5

As can be seen from the result, when add(1, 2) is called the second time, the function body is not actually executed; instead, the cached result is returned directly.

To use lru_cahce in Python 2, you need to install the third-party module functools32. There is also a faster third-party module implemented in C that is compatible with both Python2 and Python3, fastcache, which can achieve the same functionality and also supports TTL.

lru_cahce caches data in memory; in fact, data can also be cached on disk. The following example attempts to implement a disk-based cache decorator:

import os
import uuid
import pickle
import shutil
import tempfile
from functools import wraps as func_wraps

class DiskCache(object):
    """缓存数据到磁盘

    实例化参数:
    -----
        cache_path: 缓存文件的路径
    """

    _NAMESPACE = uuid.UUID("c875fb30-a8a8-402d-a796-225a6b065cad")

    def __init__(self, cache_path=None):
        if cache_path:
            self.cache_path = os.path.abspath(cache_path)
        else:
            self.cache_path = os.path.join(tempfile.gettempdir(), ".diskcache")

    def __call__(self, func):
        """返回一个包装后的函数

        如果磁盘中没有缓存，则调用函数获得结果并缓存后再返回
        如果磁盘中有缓存，则直接返回缓存的结果
        """
        @func_wraps(func)
        def wrapper(*args, **kw):
            params_uuid = uuid.uuid5(self._NAMESPACE, "-".join(map(str, (args, kw))))
            key = '{}-{}.cache'.format(func.__name__, str(params_uuid))
            cache_file = os.path.join(self.cache_path, key)

            if not os.path.exists(self.cache_path):
                os.makedirs(self.cache_path)

            try:
                with open(cache_file, 'rb') as f:
                    val = pickle.load(f)
            except Exception:
                val = func(*args, **kw)
                try:
                    with open(cache_file, 'wb') as f:
                        pickle.dump(val, f)
                except Exception:
                    pass
            return val
        return wrapper

    def clear(self, func_name):
        """清理指定函数调用的缓存"""
        for cache_file in os.listdir(self.cache_path):
            if cache_file.startswith(func_name + "-"):
                os.remove(os.path.join(self.cache_path, cache_file))

    def clear_all(self):
        """清理所有缓存"""
        if os.path.exists(self.cache_path):
            shutil.rmtree(self.cache_path)

cache_in_disk = DiskCache()

@cache_in_disk
def add(x, y):
    return x + y

In addition, there are some other caching modules, such as cachelib, cacheout, and so on. When actually needed in practice, you can choose an appropriate caching implementation according to your requirements.

Original

Original: http://kuanghy.github.io/2016/04/20/python-cache