考虑以下几点:
@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,真正的计算不依赖于可变值
当前回答
fastcache,这是Python 3 functools.lru_cache的C实现。提供10-30倍于标准库的加速。”
和选择的答案一样,只是导入不同:
from fastcache import lru_cache
@lru_cache(maxsize=128, typed=False)
def f(a, b):
pass
此外,它安装在Anaconda中,不像functools需要安装。
其他回答
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)
免责声明:我是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))来装饰你的函数/属性/类/方法…)
在Python Wiki中还有另一个备忘录装饰器的例子:
http://wiki.python.org/moin/PythonDecoratorLibrary#Memoize
这个例子有点聪明,因为如果参数是可变的,它不会缓存结果。(检查代码,它非常简单和有趣!)
函数缓存简单解决方案
TTL(时间到生命)和max_entries
当修饰函数接受不可哈希类型作为输入(例如dicts)时,不工作 可选参数:TTL(每个条目的生存时间) 可选参数:max_entries(如果缓存参数组合太多,不会使存储混乱) 确保该函数没有重要的副作用
示例使用
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
尝试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