Python 注解方式實現(xiàn)緩存數(shù)據(jù)詳解
發(fā)布日期：2021-12-23 09:32 | 文章來源：源碼中國
背景
每次加載數(shù)據(jù)都要重新Load，想通過加入的注解方式開發(fā)緩存機(jī)制，每次緩存不用寫代碼了
缺點：目前僅支持一個返回值，雖然能弄成字典，但是已經(jīng)滿足個人需求，沒動力改(狗頭)。
拿來即用
新建文件 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類就可以啦，但是有很多局限，例如只能指定某個參數(shù)被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 函數(shù)裝飾器
Python函數(shù)屬性和PyCodeObject
總結(jié)
本篇文章就到這里了，希望能夠給你帶來幫助，也希望您能夠多多關(guān)注本站的更多內(nèi)容！
版權(quán)聲明：本站文章來源標(biāo)注為YINGSOO的內(nèi)容版權(quán)均為本站所有，歡迎引用、轉(zhuǎn)載，請保持原文完整并注明來源及原文鏈接。禁止復(fù)制或仿造本網(wǎng)站，禁止在非www.sddonglingsh.com所屬的服務(wù)器上建立鏡像，否則將依法追究法律責(zé)任。本站部分內(nèi)容來源于網(wǎng)友推薦、互聯(lián)網(wǎng)收集整理而來，僅供學(xué)習(xí)參考，不代表本站立場，如有內(nèi)容涉嫌侵權(quán)，請聯(lián)系alex-e#qq.com處理。
相關(guān)文章
關(guān)注官方微信