怎么用Python注解方式實現(xiàn)緩存數(shù)據(jù)詳解

這篇文章主要介紹“怎么用Python注解方式實現(xiàn)緩存數(shù)據(jù)詳解”，在日常操作中，相信很多人在怎么用Python注解方式實現(xiàn)緩存數(shù)據(jù)詳解問題上存在疑惑，小編查閱了各式資料，整理出簡單好用的操作方法，希望對大家解答”怎么用Python注解方式實現(xiàn)緩存數(shù)據(jù)詳解”的疑惑有所幫助！接下來，請跟著小編一起來學習吧！

背景

每次加載數(shù)據(jù)都要重新Load，想通過加入的注解方式開發(fā)緩存機制，每次緩存不用寫代碼了

缺點：目前僅支持一個返回值，雖然能弄成字典，但是已經(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

到此，關(guān)于“怎么用Python注解方式實現(xiàn)緩存數(shù)據(jù)詳解”的學習就結(jié)束了，希望能夠解決大家的疑惑。理論與實踐的搭配能更好的幫助大家學習，快去試試吧！若想繼續(xù)學習更多相關(guān)知識，請繼續(xù)關(guān)注億速云網(wǎng)站，小編會繼續(xù)努力為大家?guī)砀鄬嵱玫奈恼拢?/p>