memoize to disk - python - persistent memoization

Question

有没有办法将函数的输出记录到磁盘上？

我有一个函数

def getHtmlOfUrl(url):
    ... # expensive computation

并且想要做一些类似的事情：

def getHtmlMemoized(url) = memoizeToFile(getHtmlOfUrl, "file.dat")

然后调用getHtmlMemoized( url )，以便对每个url仅执行一次昂贵的计算。

Answer 1

Python提供了一种非常优雅的方式--装饰器。基本上，装饰器是一个函数，它包装另一个函数，在不更改函数源代码的情况下提供附加功能。你的装饰器可以写成这样：

import json

def persist_to_file(file_name):

    def decorator(original_func):

        try:
            cache = json.load(open(file_name, 'r'))
        except (IOError, ValueError):
            cache = {}

        def new_func(param):
            if param not in cache:
                cache[param] = original_func(param)
                json.dump(cache, open(file_name, 'w'))
            return cache[param]

        return new_func

    return decorator

一旦你得到了它，使用@-语法“修饰”这个函数，你就准备好了。

@persist_to_file('cache.dat')
def html_of_url(url):
    your function code...

请注意，这个修饰器经过了有意的简化，可能并不适用于所有情况，例如，当源函数接受或返回不能被json序列化的数据时。

有关装饰器的更多信息：How to make a chain of function decorators?

下面是如何让装饰器在退出时只保存一次缓存：

import json, atexit

def persist_to_file(file_name):

    try:
        cache = json.load(open(file_name, 'r'))
    except (IOError, ValueError):
        cache = {}

    atexit.register(lambda: json.dump(cache, open(file_name, 'w')))

    def decorator(func):
        def new_func(param):
            if param not in cache:
                cache[param] = func(param)
            return cache[param]
        return new_func

    return decorator

Answer 2

查看joblib.Memory。它就是一个可以做到这一点的库。

from joblib import Memory
memory = Memory("cachedir")
@memory.cache
def f(x):
    print('Running f(%s)' % x)
    return x

Answer 3

一个由Python的Shelve module提供支持的更干净的解决方案。优点是缓存通过众所周知的dict语法实时更新，而且它是异常证明(不需要处理恼人的KeyError)。

import shelve
def shelve_it(file_name):
    d = shelve.open(file_name)

    def decorator(func):
        def new_func(param):
            if param not in d:
                d[param] = func(param)
            return d[param]

        return new_func

    return decorator

@shelve_it('cache.shelve')
def expensive_funcion(param):
    pass

这将使函数只需计算一次。下一次调用将返回存储的结果。