2

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

 2 years ago
source link: https://studygolang.com/articles/35277
Go to the source link to view the article. You can view the picture content, updated content and better typesetting reading experience. If the link is broken, please click the button below to view the snapshot at that time.

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

e31845690mei · 2天之前 · 128 次点击 · 预计阅读时间 5 分钟 · 大约8小时之前 开始浏览    

目录 背景 拿来即用 实践过程 通过装饰器类简化代码 总结

背景 每次加载数据都要重新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() Python客栈送红包、纸质书

只要继承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']

通过装饰器类简化代码

https://studygolang.com/tag/%E5%8D%8E%E7%BA%B3%E5%85%AC%E5%8F%B8%E6%B3%A8%E5%86%8C%E5%BC%80%E6%88%B718313230473

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


有疑问加站长微信联系(非本文作者))

280

入群交流(和以上内容无关):加入Go大咖交流群,或添加微信:liuxiaoyan-s 备注:入群;或加QQ群:701969077


About Joyk


Aggregate valuable and interesting links.
Joyk means Joy of geeK