考虑以下几点:
@property
def name(self):
if not hasattr(self, '_name'):
# expensive calculation
self._name = 1 + 1
return self._name
我是新来的,但我认为缓存可以分解成一个装饰器。只是我没有找到一个这样的;)
PS,真正的计算不依赖于可变值
考虑以下几点:
@property
def name(self):
if not hasattr(self, '_name'):
# expensive calculation
self._name = 1 + 1
return self._name
我是新来的,但我认为缓存可以分解成一个装饰器。只是我没有找到一个这样的;)
PS,真正的计算不依赖于可变值
当前回答
免责声明:我是kids.cache的作者。
你应该检查孩子。Cache,它提供了一个在python 2和python 3上工作的@cache装饰器。没有依赖关系,大约100行代码。它的使用非常简单,例如,在你的代码中,你可以这样使用:
pip install kids.cache
Then
from kids.cache import cache
...
class MyClass(object):
...
@cache # <-- That's all you need to do
@property
def name(self):
return 1 + 1 # supposedly expensive calculation
或者你可以把@cache装饰器放在@属性之后(同样的结果)。
在属性上使用缓存被称为惰性求值。缓存可以做更多的事情(它适用于带有任何参数、属性、任何类型的方法,甚至是类的函数……)对于高级用户,儿童。cache支持cachetools,它为python 2和python 3提供了漂亮的缓存存储(LRU, LFU, TTL, RR缓存)。
重要提示:孩子的默认缓存存储。缓存是一个标准字典,不建议长时间运行具有不同查询的程序,因为它会导致缓存存储不断增长。对于这种用法,你可以使用插件其他缓存存储使用例如(@cache(use=cachetools.LRUCache(maxsize=2))来装饰你的函数/属性/类/方法…)
其他回答
我编写了这个简单的装饰器类来缓存函数响应。我发现它对我的项目非常有用:
from datetime import datetime, timedelta
class cached(object):
def __init__(self, *args, **kwargs):
self.cached_function_responses = {}
self.default_max_age = kwargs.get("default_cache_max_age", timedelta(seconds=0))
def __call__(self, func):
def inner(*args, **kwargs):
max_age = kwargs.get('max_age', self.default_max_age)
if not max_age or func not in self.cached_function_responses or (datetime.now() - self.cached_function_responses[func]['fetch_time'] > max_age):
if 'max_age' in kwargs: del kwargs['max_age']
res = func(*args, **kwargs)
self.cached_function_responses[func] = {'data': res, 'fetch_time': datetime.now()}
return self.cached_function_responses[func]['data']
return inner
用法很简单:
import time
@cached
def myfunc(a):
print "in func"
return (a, datetime.now())
@cached(default_max_age = timedelta(seconds=6))
def cacheable_test(a):
print "in cacheable test: "
return (a, datetime.now())
print cacheable_test(1,max_age=timedelta(seconds=5))
print cacheable_test(2,max_age=timedelta(seconds=5))
time.sleep(7)
print cacheable_test(3,max_age=timedelta(seconds=5))
听起来好像您不是在要求一个通用的记忆化装饰器(也就是说,您对想要缓存不同参数值的返回值的一般情况不感兴趣)。也就是说,你想要这样:
x = obj.name # expensive
y = obj.name # cheap
而一个通用的记忆装饰器会给你这样的:
x = obj.name() # expensive
y = obj.name() # cheap
我认为方法调用语法是更好的风格,因为它暗示了昂贵计算的可能性,而属性语法暗示了快速查找。
[更新:我之前链接并引用的基于类的记忆化装饰器不适用于方法。我用decorator函数替换了它。如果你愿意使用通用的记忆装饰器,这里有一个简单的:
def memoize(function):
memo = {}
def wrapper(*args):
if args in memo:
return memo[args]
else:
rv = function(*args)
memo[args] = rv
return rv
return wrapper
使用示例:
@memoize
def fibonacci(n):
if n < 2: return n
return fibonacci(n - 1) + fibonacci(n - 2)
可以在这里找到另一个对缓存大小有限制的内存装饰器。
class memorize(dict):
def __init__(self, func):
self.func = func
def __call__(self, *args):
return self[args]
def __missing__(self, key):
result = self[key] = self.func(*key)
return result
示例使用:
>>> @memorize
... def foo(a, b):
... return a * b
>>> foo(2, 4)
8
>>> foo
{(2, 4): 8}
>>> foo('hi', 3)
'hihihi'
>>> foo
{(2, 4): 8, ('hi', 3): 'hihihi'}
除了Memoize示例,我还找到了以下python包:
cachepy;它允许设置ttl和\或缓存函数的调用次数;此外,还可以使用加密的基于文件的缓存… percache
from functools import wraps
def cache(maxsize=128):
cache = {}
def decorator(func):
@wraps(func)
def inner(*args, no_cache=False, **kwargs):
if no_cache:
return func(*args, **kwargs)
key_base = "_".join(str(x) for x in args)
key_end = "_".join(f"{k}:{v}" for k, v in kwargs.items())
key = f"{key_base}-{key_end}"
if key in cache:
return cache[key]
res = func(*args, **kwargs)
if len(cache) > maxsize:
del cache[list(cache.keys())[0]]
cache[key] = res
return res
return inner
return decorator
def async_cache(maxsize=128):
cache = {}
def decorator(func):
@wraps(func)
async def inner(*args, no_cache=False, **kwargs):
if no_cache:
return await func(*args, **kwargs)
key_base = "_".join(str(x) for x in args)
key_end = "_".join(f"{k}:{v}" for k, v in kwargs.items())
key = f"{key_base}-{key_end}"
if key in cache:
return cache[key]
res = await func(*args, **kwargs)
if len(cache) > maxsize:
del cache[list(cache.keys())[0]]
cache[key] = res
return res
return inner
return decorator
示例使用
import asyncio
import aiohttp
# Removes the aiohttp ClientSession instance warning.
class HTTPSession(aiohttp.ClientSession):
""" Abstract class for aiohttp. """
def __init__(self, loop=None) -> None:
super().__init__(loop=loop or asyncio.get_event_loop())
def __del__(self) -> None:
if not self.closed:
self.loop.run_until_complete(self.close())
self.loop.close()
return
session = HTTPSession()
@async_cache()
async def query(url, method="get", res_method="text", *args, **kwargs):
async with getattr(session, method.lower())(url, *args, **kwargs) as res:
return await getattr(res, res_method)()
async def get(url, *args, **kwargs):
return await query(url, "get", *args, **kwargs)
async def post(url, *args, **kwargs):
return await query(url, "post", *args, **kwargs)
async def delete(url, *args, **kwargs):
return await query(url, "delete", *args, **kwargs)