Python 注解方式实现缓存数据详解

背景

每次加载数据都要重新Load，想通过加入的注解方式开发缓存机制，每次缓存不用写代码了

缺点：目前仅支持一个返回值，虽然能弄成字典，但是已经满足个人需求，没动力改(狗头)。

拿来即用

新建文件 Cache.py

class Cache:
   def __init__(self, cache_path='.', nocache=False):
       self.cache_path = cache_path
       self.cache = not nocache
   def __call__(self, func):
       @wraps(func)
       def wrapper(*args, **kwargs):
           s = f'{func.__code__.co_filename}.{func.__name__}'
           s += ','.join(list(args[1:]) + [f'{k}={v}' for k, v in kwargs.items()])
           md5 = hashlib.md5()
           md5.update(s.encode('utf-8'))
           cache_file = f'{self.cache_path}/{md5.hexdigest()}'
           if self.cache and os.path.exists(cache_file):
               print('Loading from cache')
               return pickle.load(open(cache_file, 'rb'))
           else:
               if not os.path.exists(self.cache_path):
                   os.makedirs(self.cache_path)
               data = func(*args, **kwargs)
               pickle.dump(data, file=open(cache_file, 'wb'))
               print(f'Dump finished {cache_file}')
           return data
       return wrapper

from .Cache import Cache
@Cache(root_path, nocache=True)
def load_data(self, inpath):
   return 'Wula~a~a~!'

实践过程

第一次，来个简单的继承父类

class Cache(object):
   def __init__(self, cache_path=None):
       self.cache_path = cache_path if cache_path else '.'
       self.cache_path = f'{self.cache_path}/cache'
       self.data = self.load_cache()
   def load_cache(self):
       if os.path.exists(self.cache_path):
           print('Loading from cache')
           return pickle.load(open(self.cache_path, 'rb'))
       else:
           return None
   def save_cache(self):
       pickle.dump(self.data, file=open(self.cache_path, 'wb'))
       print(f'Dump finished {self.cache_path}')
class Filter4Analyzer(Cache):
   def __init__(self, rootpath, datapath):
       super().__init__(rootpath)
       self.root_path = rootpath
       if self.data is None:
           self.data = self.load_data(datapath)
           self.save_cache()

只要继承Cache类就可以啦，但是有很多局限，例如只能指定某个参数被cache，例如还得在Filter4Analyzer里面写保存的代码。

下一步，python嵌套装饰器来改善这个问题

from functools import wraps
import hashlib
def cached(cache_path):
   def wrapperper(func):
       @wraps(func)
       def wrapper(*args, **kwargs):
           s = f'{func.__code__.co_filename}.{func.__name__}' + ','.join(args[1:])
           s += ','.join(list(args[1:]) + [f'{k}={v}' for k, v in kwargs.items()])
           md5 = hashlib.md5()
           md5.update(s.encode('utf-8'))
           cache_file = f'{cache_path}/{md5.hexdigest()}' if cache_path else './cache'
           if os.path.exists(cache_file):
               print('Loading from cache')
               return pickle.load(open(cache_file, 'rb'))
           else:
               if not os.path.exists(cache_path):
                   os.makedirs(cache_path)
               data = func(*args, **kwargs)
               pickle.dump(data, file=open(cache_file, 'wb'))
               print(f'Dump finished {cache_file}')
           return data
       return wrapper
   return wrapperper
class Tester:
   @cached(cache_path='./workpath_test')
   def test(self, data_path):
       return ['hiahia']

通过装饰器类简化代码

class Cache:
   def __init__(self, cache_path='.', nocache=False):
       self.cache_path = cache_path
       self.cache = not nocache
   def __call__(self, func):
       @wraps(func)
       def wrapper(*args, **kwargs):
           s = f'{func.__code__.co_filename}.{func.__name__}'
           s += ','.join(list(args[1:]) + [f'{k}={v}' for k, v in kwargs.items()])
           md5 = hashlib.md5()
           md5.update(s.encode('utf-8'))
           cache_file = f'{self.cache_path}/{md5.hexdigest()}'
           if self.cache and os.path.exists(cache_file):
               print('Loading from cache')
               return pickle.load(open(cache_file, 'rb'))
           else:
               if not os.path.exists(self.cache_path):
                   os.makedirs(self.cache_path)
               data = func(*args, **kwargs)
               pickle.dump(data, file=open(cache_file, 'wb'))
               print(f'Dump finished {cache_file}')
           return data
       return wrapper

参考：

Python 函数装饰器

Python函数属性和PyCodeObject

来源：https://blog.csdn.net/liuxing93619/article/details/120837018