考虑以下:
@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真正的计算不依赖于可变值
从 Python 3.2 开始,有一个内置的装饰器:
@functools.lru_cache(maxsize=100, typed=False)
装饰器用一个可保存最多 maxsize 最近调用的记忆可调用函数来包装函数。当使用相同的参数定期调用昂贵的或 I/O 绑定的函数时,它可以节省时间。
用于计算斐波那契数的 LRU 缓存示例:
from functools import lru_cache
@lru_cache(maxsize=None)
def fib(n):
if n < 2:
return n
return fib(n-1) + fib(n-2)
>>> print([fib(n) for n in range(16)])
[0, 1, 1, 2, 3, 5, 8, 13, 21, 34, 55, 89, 144, 233, 377, 610]
>>> print(fib.cache_info())
CacheInfo(hits=28, misses=16, maxsize=None, currsize=16)
如果你被 Python 2.x 卡住了,这里是其他兼容的记忆库的列表:
听起来您不是在要求通用的记忆装饰器(即,您对要缓存不同参数值的返回值的一般情况不感兴趣)。也就是说,你想要这个:
x = obj.name # expensive
y = obj.name # cheap
而一个通用的记忆装饰器会给你这个:
x = obj.name() # expensive
y = obj.name() # cheap
我认为方法调用语法是更好的风格,因为它暗示了计算成本高昂的可能性,而属性语法则暗示了快速查找。
[更新:我之前在这里链接和引用的基于类的记忆装饰器不适用于方法。我已经用装饰器函数替换了它。] 如果你愿意使用通用的记忆装饰器,这里有一个简单的:
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)
可以在此处找到另一个对缓存大小有限制的记忆装饰器。
Python 3.8functools.cached_property
装饰器
https://docs.python.org/dev/library/functools.html#functools.cached_property
cached_property
来自 Werkzeug 的内容在:https ://stackoverflow.com/a/5295190/895245 中被提及,但据说派生的版本将合并到 3.8 中,这太棒了。
这个装饰器可以被看作是缓存,或者当你没有任何参数时@property
作为一个清洁器 。@functools.lru_cache
文档说:
@functools.cached_property(func)
将类的方法转换为属性,其值被计算一次,然后在实例的生命周期内作为普通属性缓存。与 property() 类似,但增加了缓存。对于实例的昂贵计算属性很有用,否则这些属性实际上是不可变的。
例子:
class DataSet: def __init__(self, sequence_of_numbers): self._data = sequence_of_numbers @cached_property def stdev(self): return statistics.stdev(self._data) @cached_property def variance(self): return statistics.variance(self._data)
3.8 版中的新功能。
注意此装饰器要求每个实例上的dict属性是可变映射。这意味着它不适用于某些类型,例如元类(因为类型实例上的dict属性是类名称空间的只读代理),以及那些指定插槽而不包括dict作为已定义插槽之一的类型(如此类根本不提供dict属性)。
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'}
functools.cache
已在 Python 3.9 ( docs ) 中发布:
from functools import cache
@cache
def factorial(n):
return n * factorial(n-1) if n else 1
在以前的 Python 版本中,早期的答案之一仍然是lru_cache
作为没有限制和 lru 功能的普通缓存使用的有效解决方案。(文档)
如果 maxsize 设置为 None,则禁用 LRU 功能,并且缓存可以无限制地增长。
这是它的一个更漂亮的版本:
cache = lru_cache(maxsize=None)
@cache
def func(param1):
pass
我编写了这个简单的装饰器类来缓存函数响应。我发现它对我的项目非常有用:
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))
啊,只需要为此找到正确的名称:“惰性属性评估”。
我也经常这样做;也许我会在某个时候在我的代码中使用那个食谱。
免责声明:我是kids.cache的作者。
你应该检查一下kids.cache
,它提供了一个@cache
适用于 python 2 和 python 3 的装饰器。没有依赖关系,大约 100 行代码。它使用起来非常简单,例如,考虑到您的代码,您可以像这样使用它:
pip install kids.cache
然后
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
装饰器放在@property
(相同的结果)之后。
在属性上使用缓存称为惰性评估,kids.cache
可以做更多事情(它适用于具有任何参数、属性、任何类型的方法甚至类的函数......)。对于高级用户,kids.cache
支持cachetools
为 python 2 和 python 3(LRU、LFU、TTL、RR 缓存)提供精美的缓存存储。
重要说明:默认的缓存存储kids.cache
是标准字典,不建议将其用于具有不同查询的长时间运行的程序,因为它会导致缓存存储不断增长。对于这种用法,您可以使用例如插件其他缓存存储(@cache(use=cachetools.LRUCache(maxsize=2))
以装饰您的函数/属性/类/方法...)
Python Wiki上还有另一个memoize装饰器示例:
http://wiki.python.org/moin/PythonDecoratorLibrary#Memoize
这个例子有点聪明,因为如果参数是可变的,它就不会缓存结果。(检查那个代码,它非常简单有趣!)
如果您使用的是 Django 框架,它有这样一个属性来缓存 API 使用的视图或响应,@cache_page(time)
并且还可以有其他选项。
例子:
@cache_page(60 * 15, cache="special_cache")
def my_view(request):
...
更多细节可以在这里找到。
试试joblib https://joblib.readthedocs.io/en/latest/memory.html
from joblib import Memory
memory = Memory(cachedir=cachedir, verbose=0)
@memory.cache
def f(x):
print('Running f(%s)' % x)
return x
除了Memoize 示例,我还发现了以下 python 包:
@lru_cache
默认属性不好
我的@mem
装饰师:
import inspect
from copy import deepcopy
from functools import lru_cache, wraps
from typing import Any, Callable, Dict, Iterable
# helper
def get_all_kwargs_values(f: Callable, kwargs: Dict[str, Any]) -> Iterable[Any]:
default_kwargs = {
k: v.default
for k, v in inspect.signature(f).parameters.items()
if v.default is not inspect.Parameter.empty
}
all_kwargs = deepcopy(default_kwargs)
all_kwargs.update(kwargs)
for key in sorted(all_kwargs.keys()):
yield all_kwargs[key]
# the best decorator
def mem(func: Callable) -> Callable:
cache = dict()
@wraps(func)
def wrapper(*args, **kwargs) -> Any:
all_kwargs_values = get_all_kwargs_values(func, kwargs)
params = (*args, *all_kwargs_values)
_hash = hash(params)
if _hash not in cache:
cache[_hash] = func(*args, **kwargs)
return cache[_hash]
return wrapper
# some logic
def counter(*args) -> int:
print(f'* not_cached:', end='\t')
return sum(args)
@mem
def check_mem(a, *args, z=10) -> int:
return counter(a, *args, z)
@lru_cache
def check_lru(a, *args, z=10) -> int:
return counter(a, *args, z)
def test(func) -> None:
print(f'\nTest {func.__name__}:')
print('*', func(1, 2, 3, 4, 5))
print('*', func(1, 2, 3, 4, 5))
print('*', func(1, 2, 3, 4, 5, z=6))
print('*', func(1, 2, 3, 4, 5, z=6))
print('*', func(1))
print('*', func(1, z=10))
def main():
test(check_mem)
test(check_lru)
if __name__ == '__main__':
main()
输出:
Test check_mem:
* not_cached: * 25
* 25
* not_cached: * 21
* 21
* not_cached: * 11
* 11
Test check_lru:
* not_cached: * 25
* 25
* not_cached: * 21
* 21
* not_cached: * 11
* not_cached: * 11
我实现了类似的东西,使用 pickle 来实现持久性,并使用 sha1 来实现几乎肯定唯一的简短 ID。基本上,缓存对函数的代码和参数的历史进行哈希处理以获取 sha1,然后查找名称中包含该 sha1 的文件。如果存在,则打开它并返回结果;如果没有,它会调用该函数并保存结果(可选地,仅在需要一定时间来处理时才保存)。
也就是说,我发誓我找到了一个执行此操作的现有模块,并发现自己在这里试图找到该模块......我能找到的最接近的是这个,看起来很正确:http://chase-seibert.github。 io/blog/2011/11/23/pythondjango-disk-based-caching-decorator.html
我看到的唯一问题是它不适用于大型输入,因为它散列 str(arg),这对于巨型数组来说并不是唯一的。
如果有一个unique_hash () 协议让一个类返回其内容的安全散列,那就太好了。对于我关心的类型,我基本上是手动实现的。
如果您使用 Django 并想要缓存视图,请参阅Nikhil Kumar 的回答。
但是如果你想缓存任何函数结果,你可以使用django-cache-utils。
它重用 Django 缓存并提供易于使用的cached
装饰器:
from cache_utils.decorators import cached
@cached(60)
def foo(x, y=0):
print 'foo is called'
return x+y
import time
@cache(ttl=timedelta(minutes=3), max_entries=300)
def add(a, b):
time.sleep(2)
return a + b
@cache()
def substract(a, b):
time.sleep(2)
return a - b
a = 5
# function is called with argument combinations the first time -> it takes some time
for i in range(5):
print(add(a, i))
# function is called with same arguments again? -> will answer from cache
for i in range(5):
print(add(a, i))
from datetime import datetime, timedelta
def cache(**kwargs):
def decorator(function):
# static function variable for cache, lazy initialization
try: function.cache
except: function.cache = {}
def wrapper(*args):
# if nothing valid in cache, insert something
if not args in function.cache or datetime.now() > function.cache[args]['expiry']:
if 'max_entries' in kwargs:
max_entries = kwargs['max_entries']
if max_entries != None and len(function.cache) >= max_entries:
now = datetime.now()
# delete the the first expired entry that can be found (lazy deletion)
for key in function.cache:
if function.cache[key]['expiry'] < now:
del function.cache[key]
break
# if nothing is expired that is deletable, delete the first
if len(function.cache) >= max_entries:
del function.cache[next(iter(function.cache))]
function.cache[args] = {'result': function(*args), 'expiry': datetime.max if 'ttl' not in kwargs else datetime.now() + kwargs['ttl']}
# answer from cache
return function.cache[args]['result']
return wrapper
return decorator
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)