背景
每次加載數(shù)據(jù)都要重新Load，想通過加入的注解方式開發(fā)緩存機(jī)制，每次緩存不用寫代碼了
缺點(diǎn)：目前僅支持一個(gè)返回值，雖然能弄成字典，但是已經(jīng)滿足個(gè)人需求，沒動(dòng)力改(狗頭)。
拿來即用
新建文件 Cache.py
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
|
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 |
|
1
2
3
4
|
from .Cache import Cache@Cache(root_path, nocache=True)def load_data(self, inpath): return 'Wula~a~a~!' |
實(shí)踐過程
第一次，來個(gè)簡單的繼承父類
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
|
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類就可以啦，但是有很多局限，例如只能指定某個(gè)參數(shù)被cache，例如還得在Filter4Analyzer里面寫保存的代碼。
下一步，python嵌套裝飾器來改善這個(gè)問題
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
|
from functools import wrapsimport hashlibdef 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 wrapperperclass Tester: @cached(cache_path='./workpath_test') def test(self, data_path): return ['hiahia'] |
通過裝飾器類簡化代碼
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
|
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ù)屬性和PyCodeObject
總結(jié)
本篇文章就到這里了，希望能夠給你帶來幫助，也希望您能夠多多關(guān)注服務(wù)器之家的更多內(nèi)容！
原文鏈接：https://blog.csdn.net/liuxing93619/article/details/120837018
