是否有一个装饰器可以简单地缓存函数返回值?

2025-02-14 09:50:00
admin
原创
73
摘要:问题描述:请考虑以下情况:@property def name(self): if not hasattr(self, '_name'): # expensive calculation self._name = 1 + 1 return self._na...

问题描述:

请考虑以下情况:

@property
def name(self):

    if not hasattr(self, '_name'):

        # expensive calculation
        self._name = 1 + 1

    return self._name

我是新手,但我认为可以将缓存功能分解为装饰器。只是我没有找到类似的装饰器 ;)

PS 实际计算不依赖于可变值


解决方案 1:

从 Python 3.2 开始有一个内置的装饰器:

@functools.lru_cache(maxsize=100, typed=False)

装饰器用一个可记忆调用函数包装一个函数,该函数最多可保存最近调用的最大数量。当使用相同参数定期调用昂贵或 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,这里是其他兼容的记忆库的列表:

  • functools32| PyPI |源代码

  • repoze.lru| PyPI |源代码

  • pylru| PyPI |源代码

  • backports.functools_lru_cache| PyPI |源代码

解决方案 2:

functools.cache已在 Python 3.9 中发布(文档):

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

解决方案 3:

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属性)。

解决方案 4:

听起来你并不是在要求一个通用的记忆化装饰器(即,你对想要缓存不同参数值的返回值的一般情况不感兴趣)。也就是说,你想要的是这样的:

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)

可以在此处找到另一个对缓存大小有限制的记忆装饰器。

解决方案 5:

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'}

解决方案 6:

Werkzeug 有一个cached_property装饰器(文档,源代码)

解决方案 7:

我编写了这个简单的装饰器类来缓存函数响应。我发现它对我的项目非常有用:

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))

解决方案 8:

尝试 joblib
https://joblib.readthedocs.io/en/latest/memory.html

from joblib import Memory

# customize the decorator
memory = Memory(cachedir=cachedir, verbose=0)

@memory.cache
def f(x):
    print('Running f(%s)' % x)
    return x

解决方案 9:

免责声明:我是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))装饰您的函数/属性/类/方法...)

解决方案 10:

啊,只需要为此找到正确的名称:“惰性属性评估”。

我也经常这样做;也许我有时会在我的代码中使用这个方法。

解决方案 11:

Python Wiki 上还有另一个memoize装饰器的示例:

http://wiki.python.org/moin/PythonDecoratorLibrary#Memoize

这个例子有点聪明,因为如果参数是可变的,它就不会缓存结果。(检查一下代码,它非常简单而且有趣!)

解决方案 12:

如果您正在使用 Django 框架,它具有这样的属性来缓存 API 使用的视图或响应@cache_page(time),并且还可以有其他选项。

例子:

@cache_page(60 * 15, cache="special_cache")
def my_view(request):
    ...

更多详细信息请参见此处。

解决方案 13:

有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 则需要另行安装。

解决方案 14:

除了Memoize 示例之外,我还发现了以下 Python 包:

  • cachepy;它允许设置 ttl 和\或缓存函数的调用次数;此外,还可以使用基于加密文件的缓存......

  • 缓存

解决方案 15:

@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='    ')
    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'
Test {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

解决方案 16:

创建自己的装饰器并使用它

from django.core.cache import cache
import functools

def cache_returned_values(func):
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        key = "choose a unique key here"
        results = cache.get(key)
        if not results:
            results = func(*args, **kwargs)
            cache.set(key, results)
        return results

    return wrapper

现在在函数方面

@cache_returned_values
def get_some_values(args):
  return x

解决方案 17:

我实现了类似的东西,使用 pickle 进行持久化,使用 sha1 获得简短且几乎肯定唯一的 ID。基本上,缓存对函数的代码和参数的 hist 进行哈希处理以获取 sha1,然后查找名称中包含该 sha1 的文件。如果存在,它会打开它并返回结果;如果不存在,它会调用该函数并保存结果(可选,仅在处理需要一定时间时才保存)。

话虽如此,我发誓我找到了一个可以做到这一点的现有模块,并且我在这里试图找到该模块...我能找到的最接近的是这个,看起来正确:http://chase-seibert.github.io/blog/2011/11/23/pythondjango-disk-based-caching-decorator.html

我看到的唯一问题是它不适用于大型输入,因为它对 str(arg) 进行散列,而这对于大型数组来说并不唯一。

如果有一个unique_hash () 协议,让类返回其内容的安全哈希值,那就太好了。我基本上是手动为我关心的类型实现的。

解决方案 18:

如果您正在使用 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

解决方案 19:

函数缓存简单解决方案

使用 ttl (有效期) 和 max_entries

  • 当修饰函数采用不可散列类型作为输入(例如字典)时不起作用

  • 可选参数: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

解决方案 20:

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)
相关推荐
  政府信创国产化的10大政策解读一、信创国产化的背景与意义信创国产化,即信息技术应用创新国产化,是当前中国信息技术领域的一个重要发展方向。其核心在于通过自主研发和创新,实现信息技术应用的自主可控,减少对外部技术的依赖,并规避潜在的技术制裁和风险。随着全球信息技术竞争的加剧,以及某些国家对中国在科技领域的打压,信创国产化显...
工程项目管理   3998  
  为什么项目管理通常仍然耗时且低效?您是否还在反复更新电子表格、淹没在便利贴中并参加每周更新会议?这确实是耗费时间和精力。借助软件工具的帮助,您可以一目了然地全面了解您的项目。如今,国内外有足够多优秀的项目管理软件可以帮助您掌控每个项目。什么是项目管理软件?项目管理软件是广泛行业用于项目规划、资源分配和调度的软件。它使项...
项目管理软件   2749  
  本文介绍了以下10款项目管理软件工具:禅道项目管理软件、Freshdesk、ClickUp、nTask、Hubstaff、Plutio、Productive、Targa、Bonsai、Wrike。在当今快速变化的商业环境中,项目管理已成为企业成功的关键因素之一。然而,许多企业在项目管理过程中面临着诸多痛点,如任务分配不...
项目管理系统   85  
  本文介绍了以下10款项目管理软件工具:禅道项目管理软件、Monday、TeamGantt、Filestage、Chanty、Visor、Smartsheet、Productive、Quire、Planview。在当今快速变化的商业环境中,项目管理已成为企业成功的关键因素之一。然而,许多项目经理和团队在管理复杂项目时,常...
开源项目管理工具   96  
  本文介绍了以下10款项目管理软件工具:禅道项目管理软件、Smartsheet、GanttPRO、Backlog、Visor、ResourceGuru、Productive、Xebrio、Hive、Quire。在当今快节奏的商业环境中,项目管理已成为企业成功的关键因素之一。然而,许多企业在选择项目管理工具时常常面临困惑:...
项目管理系统   83  
热门文章
项目管理软件有哪些?
曾咪二维码

扫码咨询,免费领取项目管理大礼包!

云禅道AD
禅道项目管理软件

云端的项目管理软件

尊享禅道项目软件收费版功能

无需维护,随时随地协同办公

内置subversion和git源码管理

每天备份,随时转为私有部署

免费试用