Python使用类似于模拟补丁的技术缓存内部调用

Question

我想将缓存用于 API 中的特定功能。我不想逐行修改内部代码，而是想通过使用类似于模拟补丁的技术来实现相同的目的。例如

@cache_patch('lib.Someobjclass.func1',ttl=200)
@cache_patch('lib.Someotherobjclass.func2',ttl=1000)
function abc(*args, **kwargs):
    '''do stuff'''
    cache1 = someobj.func1(args,kwargs)
    '''do more stuff'''
    cache2 = someotherobj.func2(args,kwargs)

有没有可以使用的库或技术？

Answer 1

假设

我不是 100% 清楚你想要什么确切的行为，所以我假设缓存只在应用装饰器的函数执行期间使用。

---

必要的点点滴滴

你需要

• 使用functools.lru_cache实现缓存
• 创建一个能够接受额外参数的装饰器
• 使用importlib 将作为第一个参数给定的类从字符串导入装饰器
• 猴子用缓存版本修补所需的方法

---

把它放在一起

import importlib
from functools import lru_cache


class Addition:
    def __init__(self, a):
        self.a = a

    def uncached_addition(self, b):
        # print is only used to demonstrate if the method is actually called or not
        print(f"Computing {self.a} + {b}")
        return self.a + b


class cache_patch:
    def __init__(self, method_as_str, ttl):
        # split the given path into module, class and method name
        class_as_str, method_name = method_as_str.rsplit(".", 1)
        module_path, class_name = class_as_str.rsplit(".", 1)

        self.clazz = getattr(importlib.import_module(module_path), class_name)
        self.method_name = method_name
        self.ttl = ttl

    def __call__(self, func):
        def wrapped(*args, **kwargs):
            # monkey patch the original method with a cached version
            uncached_method = getattr(self.clazz, self.method_name)
            cached_method = lru_cache(maxsize=self.ttl)(uncached_method)
            setattr(self.clazz, self.method_name, cached_method)

            result = func(*args, **kwargs)

            # replace cached method with original
            setattr(self.clazz, self.method_name, uncached_method)
            return result
        return wrapped


@cache_patch('__main__.Addition.uncached_addition', ttl=128)
def perform_patched_uncached_addition(a, b):
    d = Addition(a=1)
    print("Patched nr. 1\t", d.uncached_addition(2))
    print("Patched nr. 2\t", d.uncached_addition(2))
    print("Patched nr. 3\t", d.uncached_addition(2))
    print()


if __name__ == '__main__':
    perform_patched_uncached_addition(1, 2)
    d = Addition(a=1)
    print("Unpatched nr. 1\t", d.uncached_addition(2))
    print("Unpatched nr. 2\t", d.uncached_addition(2))
    print("Unpatched nr. 3\t", d.uncached_addition(2))

---

结果

从输出结果可以看出，调用perform_patched_uncached_addition只会输出一次Computing 1 + 2，之后使用缓存的结果:

Computing 1 + 2
Patched call nr. 1:  3
Patched call nr. 2:  3
Patched call nr. 3:  3

在此函数之外对该类进行的任何调用都将使用该方法的未修补、非缓存版本:

Computing 1 + 2
Unpatched call nr. 1:    3
Computing 1 + 2
Unpatched call nr. 2:    3
Computing 1 + 2
Unpatched call nr. 3:    3

---

注意事项

如果你打算在多线程环境中使用这种方法，你当然需要注意。您将无法确定何时应用缓存补丁。
此外，缓存是根据传递给方法的参数完成的，其中包括 self。这意味着该类的每个实例都有自己的“缓存”。

---

旁注

与上面的代码不同，functools.lru_cache在大多数情况下被用作装饰器:

class Addition:
    def __init__(self, a):
        self.a = a

    @lru_cache(maxsize=128)
    def cached_addition(self, b):
        print(f"Computing {self.a} + b")
        return self.a + b