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

Chloe

发布时间：2026-10-14| 6人浏览过

原创

本文介绍如何在 Python 中通过注解方式实现数据缓存机制，避免每次加载数据时重新执行 Load 操作。文章从基础的 Cache 类继承方案入手，分析其局限性，随后通过嵌套装饰器类进行优化，简化代码实现。内容涵盖缓存路径配置、MD5 哈希文件名生成、pickle 序列化存储及加载逻辑，并提供了完整的代码示例与注意事项。

背景介绍

在数据处理过程中，每次加载数据都需要重新执行 Load 操作，效率较低。为了解决这一问题，希望通过引入注解（Annotation）方式开发一套缓存机制，使得在后续使用中无需编写额外的缓存处理代码，从而简化开发流程。

需要注意的是，当前实现的方案存在一定局限性：目前仅支持单个返回值的缓存。尽管技术上可以扩展为支持字典等复杂结构，但由于当前功能已满足个人使用需求，暂无进一步修改的动力。

快速上手指南

要使用此功能，只需新建一个名为 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 类即可启用缓存，但该方法存在诸多局限。例如，只能指定特定参数进行缓存，且还需要在 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