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

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

我实现了类似的东西，使用pickle进行持久化，使用sha1进行简短的几乎唯一的id。基本上，缓存对函数代码和参数的历史进行哈希，以获得sha1，然后查找名称为sha1的文件。如果它存在，则打开它并返回结果;如果没有，则调用该函数并保存结果(如果需要一定时间来处理，则可以选择只保存结果)。

也就是说，我发誓我找到了一个现有的模块，它做到了这一点，并发现自己在这里试图找到该模块……我能找到的最接近的是这个，看起来差不多:http://chase-seibert.github.io/blog/2011/11/23/pythondjango-disk-based-caching-decorator.html

我看到的唯一问题是，它不能很好地用于大输入，因为它散列str(arg)，这不是唯一的大型数组。

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

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

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

如果你正在使用Django并且想要缓存视图，请参阅Nikhil Kumar的回答。

但是如果你想缓存任何函数的结果，你可以使用django-cache-utils。

它重用了Django缓存，并提供了易于使用的缓存装饰器:

from cache_utils.decorators import cached

@cached(60)
def foo(x, y=0):
    print 'foo is called'
    return x+y

Werkzeug有一个cached_property装饰器(docs, source)

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)