Python 文件缓存

Question

我正在从文件创建一些对象（来自模板 xsd 文件的验证器，以便将其他 xsd 文件组合在一起，当它发生时），我想在磁盘上的文件更改时重新创建这些对象。

我可以创建类似的东西：

def getobj(fname, cache = {}):
    try:
        obj, lastloaded = cache[fname]
        if lastloaded < last_time_written(fname):
           # same stuff as in except clause
    except KeyError:
        obj = create_from_file(fname)
        cache[fname] = (obj, currenttime)

    return obj

但是，如果存在其他人的测试代码，我更愿意使用它。是否有现有的库可以执行此类操作？

更新：我使用的是 python 2.7.1。

Answer 1

您的代码（包括缓存逻辑）看起来不错。

考虑将 cache 变量移到函数定义之外。这样就可以添加其他功能来清除或检查缓存。

如果您想查看执行类似操作的代码，请查看filecmp 模块的源代码：http://hg.python.org/cpython/file/2.7/Lib/filecmp.py 有趣的部分是stat module 如何用于确定文件是否已更改。这是签名函数：

def _sig(st):
    return (stat.S_IFMT(st.st_mode),
            st.st_size,
            st.st_mtime)

Answer 2

三个想法。

1. 使用try... except... else 获得更简洁的控制流。

2. 众所周知，文件修改时间不稳定 - 特别是，它们不一定对应于最近修改文件的时间！

3. Python 3 包含一个缓存装饰器：functools.lru_cache。这是来源。

   def lru_cache(maxsize=100):
       """Least-recently-used cache decorator.
   
       If *maxsize* is set to None, the LRU features are disabled and the cache
       can grow without bound.
   
       Arguments to the cached function must be hashable.
   
       View the cache statistics named tuple (hits, misses, maxsize, currsize) with
       f.cache_info().  Clear the cache and statistics with f.cache_clear().
       Access the underlying function with f.__wrapped__.
   
       See:  http://en.wikipedia.org/wiki/Cache_algorithms#Least_Recently_Used
   
       """
       # Users should only access the lru_cache through its public API:
       #       cache_info, cache_clear, and f.__wrapped__
       # The internals of the lru_cache are encapsulated for thread safety and
       # to allow the implementation to change (including a possible C version).
   
       def decorating_function(user_function,
                   tuple=tuple, sorted=sorted, len=len, KeyError=KeyError):
   
           hits = misses = 0
           kwd_mark = (object(),)          # separates positional and keyword args
           lock = Lock()                   # needed because ordereddicts aren't threadsafe
   
           if maxsize is None:
               cache = dict()              # simple cache without ordering or size limit
   
               @wraps(user_function)
               def wrapper(*args, **kwds):
                   nonlocal hits, misses
                   key = args
                   if kwds:
                       key += kwd_mark + tuple(sorted(kwds.items()))
                   try:
                       result = cache[key]
                       hits += 1
                   except KeyError:
                       result = user_function(*args, **kwds)
                       cache[key] = result
                       misses += 1
                   return result
           else:
               cache = OrderedDict()       # ordered least recent to most recent
               cache_popitem = cache.popitem
               cache_renew = cache.move_to_end
   
               @wraps(user_function)
               def wrapper(*args, **kwds):
                   nonlocal hits, misses
                   key = args
                   if kwds:
                       key += kwd_mark + tuple(sorted(kwds.items()))
                   try:
                       with lock:
                           result = cache[key]
                           cache_renew(key)        # record recent use of this key
                           hits += 1
                   except KeyError:
                       result = user_function(*args, **kwds)
                       with lock:
                           cache[key] = result     # record recent use of this key
                           misses += 1
                           if len(cache) > maxsize:
                               cache_popitem(0)    # purge least recently used cache entry
                   return result
   
           def cache_info():
               """Report cache statistics"""
               with lock:
                   return _CacheInfo(hits, misses, maxsize, len(cache))
   
           def cache_clear():
               """Clear the cache and cache statistics"""
               nonlocal hits, misses
               with lock:
                   cache.clear()
                   hits = misses = 0
   
           wrapper.cache_info = cache_info
           wrapper.cache_clear = cache_clear
           return wrapper
   
       return decorating_function
   

Answer 3

除非有特定原因将其用作参数，否则我会将缓存用作全局对象