# Python3魔术方法实战：用__getitem__和__setitem__打造你的专属字典类 你是否曾对Python内置的`dict`感到既爱又恨？它灵活、高效，是日常开发中不可或缺的工具。但当你需要为键值对添加一些额外的“魔法”时——比如自动将字符串键转换为小写、验证值的类型、或者为不存在的键提供一个默认值——内置字典就显得有些力不从心了。这时，你可能会开始编写一堆辅助函数，或者在每次访问字典时都加上一层包装逻辑，代码很快就变得臃肿不堪。 其实，Python早已为你准备好了更优雅的解决方案：**魔术方法**。特别是`__getitem__`和`__setitem__`，它们就像是字典行为的“后门”，允许你深度定制键值访问的每一个细节。这不仅仅是语法糖，而是一种强大的元编程能力，能让你创造出行为独特、功能强大的自定义容器。 想象一下，你可以创建一个字典，它能自动将存入的数值进行单位换算，或者能记录每一次键值访问的日志用于调试，甚至能无缝地将本地数据与远程缓存（如Redis）同步。这些听起来复杂的功能，通过合理重写`__getitem__`和`__setitem__`，都能以清晰、内聚的方式实现。本文将带你跳出对魔术方法浅尝辄止的理解，深入这两个方法的实战应用，通过一系列从简到繁的案例，手把手教你构建出能满足特定业务需求的“超级字典”。 ## 1. 理解基石：`__getitem__`与`__setitem__`的核心机制 在动手改造之前，我们必须先弄清楚这两个方法在Python对象生命周期中扮演的角色。它们都属于Python的**协议**（Protocol）的一部分。协议是一种非正式的接口，只要你的类实现了特定方法，就能支持相应的语言特性。对于字典式的访问协议，核心就是`__getitem__`和`__setitem__`。 **`__getitem__(self, key)`**：当你使用`obj[key]`这样的下标语法获取元素时，Python解释器会悄悄调用这个方法。参数`key`可以是任何对象，不仅仅是字符串或整数。这个方法需要返回与`key`对应的值，如果`key`不存在，按照惯例应该抛出`KeyError`异常。 **`__setitem__(self, key, value)`**：当你使用`obj[key] = value`进行赋值时，解释器会调用这个方法。它负责将`value`与`key`关联起来。注意，这个方法**不返回任何值**（或者说返回`None`）。 一个常见的误解是，只有继承自`dict`的类才能使用这些方法。实际上，任何类只要实现了`__getitem__`，就能支持`[]`取值操作；实现了`__setitem__`，就能支持`[]`赋值操作。这使得我们可以从零开始构建一个行为类似字典的类，或者为现有类添加字典式接口。 让我们从一个最基础的例子开始，看看如何实现一个简单的包装类： ```python class LoggingDict: """一个会记录所有访问操作的字典包装类""" def __init__(self, initial_data=None): self._data = {} if initial_data is None else dict(initial_data) def __getitem__(self, key): print(f"[GET] 正在访问键: {key!r}") # 如果键不存在，_data[key]会抛出KeyError，这正是我们想要的行为 return self._data[key] def __setitem__(self, key, value): print(f"[SET] 设置键 {key!r} 的值为: {value!r}") self._data[key] = value def __repr__(self): return f"LoggingDict({self._data})" # 使用示例 my_dict = LoggingDict({'name': 'Alice'}) my_dict['age'] = 30 # 输出: [SET] 设置键 'age' 的值为: 30 print(my_dict['name']) # 输出: [GET] 正在访问键: 'name'，然后输出: Alice print(my_dict) # 输出: LoggingDict({'name': 'Alice', 'age': 30}) ``` 这个`LoggingDict`类内部使用一个普通的`dict`（`self._data`）来存储实际数据，而`__getitem__`和`__setitem__`则充当了“拦截器”的角色。所有通过`[]`进行的操作都会先经过我们的自定义逻辑（这里只是打印日志），然后再委托给内部的`_data`字典处理。这种模式——**装饰器模式**——在自定义容器时非常常用。 > **注意**：在`__setitem__`中，我们直接操作了`self._data[key]`。为什么不使用`self._data.__setitem__(key, value)`？实际上，两者在功能上等价。但直接使用`self._data[key] = value`会再次触发`self._data`的`__setitem__`（如果它有自定义实现的话）。在这个简单例子中，`self._data`是一个普通字典，所以没有区别。但在更复杂的嵌套场景中，需要注意避免无限递归。 理解了基本机制后，我们来看一个更贴近实际需求的场景：**键的规范化**。在Web开发中，HTTP头部字段（如`Content-Type`）通常不区分大小写，但Python字典是区分的。我们可以创建一个不区分大小写的字典： ```python class CaseInsensitiveDict: def __init__(self, initial_data=None): self._data = {} # 用于存储原始键到规范化键的映射，以便在__repr__等场景中能显示原始键 self._key_mapping = {} if initial_data: for key, value in initial_data.items(): self[key] = value # 使用我们自定义的__setitem__ def _normalize_key(self, key): """将键规范化为小写字符串（可根据需要修改）""" if isinstance(key, str): return key.lower() return key # 非字符串键保持不变 def __getitem__(self, key): normalized = self._normalize_key(key) # 如果规范化后的键不存在，抛出KeyError时使用原始键更友好 if normalized not in self._data: raise KeyError(f"{key!r}") return self._data[normalized] def __setitem__(self, key, value): normalized = self._normalize_key(key) self._data[normalized] = value # 记录原始键，以便后续可能需要 self._key_mapping[normalized] = key def __delitem__(self, key): normalized = self._normalize_key(key) if normalized in self._data: del self._data[normalized] del self._key_mapping[normalized] else: raise KeyError(f"{key!r}") def __contains__(self, key): return self._normalize_key(key) in self._data def __repr__(self): # 使用原始键进行展示 items = [] for norm_key, value in self._data.items(): orig_key = self._key_mapping.get(norm_key, norm_key) items.append(f"{orig_key!r}: {value!r}") return "CaseInsensitiveDict({" + ", ".join(items) + "})" # 测试 headers = CaseInsensitiveDict({'Content-Type': 'application/json'}) headers['content-type'] = 'text/html' # 这会覆盖之前的'Content-Type' print(headers) # 输出: CaseInsensitiveDict({'content-type': 'text/html'}) print('CONTENT-TYPE' in headers) # 输出: True ``` 这个例子展示了几个重要技巧： 1. 在`__init__`中，我们使用`self[key] = value`来初始化数据，这确保了初始数据也经过键的规范化处理。 2. 我们额外维护了一个`_key_mapping`字典，用来记住每个规范化键对应的原始键。这在`__repr__`中很有用，可以让对象的字符串表示更直观。 3. 为了实现完整的字典行为，我们还需要实现`__delitem__`和`__contains__`等方法。一个完整的自定义字典通常需要实现十多个魔术方法，但`__getitem__`和`__setitem__`是最核心的。 ## 2. 进阶应用：实现数据验证与自动转换 在实际项目中，我们经常需要对字典中存储的数据进行约束。比如，确保某个键的值总是整数，或者自动将输入的字符串转换为日期对象。通过在`__setitem__`中加入验证和转换逻辑，我们可以创建出“智能”的字典，让数据在存入时就符合预期格式，从而减少后续处理中的错误。 假设我们正在开发一个配置管理系统，需要处理如下配置项： - `port`：必须是1024到65535之间的整数 - `debug`：必须是布尔值，但允许接受字符串`"true"`/`"false"` - `log_level`：必须是预定义的几种字符串之一（如`"DEBUG"`, `"INFO"`, `"WARNING"`） 我们可以创建一个`ValidatedConfigDict`类： ```python class ValidatedConfigDict: """一个对特定键进行验证和转换的配置字典""" # 定义验证规则：键名 -> (类型或可调用验证函数, 转换函数（可选）) VALIDATORS = { 'port': (lambda x: isinstance(x, int) and 1024 <= x <= 65535, int), 'debug': (lambda x: isinstance(x, bool) or x in ('true', 'false', '1', '0'), lambda x: x if isinstance(x, bool) else x.lower() in ('true', '1')), 'log_level': (lambda x: x in ('DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL'), str.upper), } def __init__(self, initial_data=None): self._data = {} if initial_data: for key, value in initial_data.items(): self[key] = value def __getitem__(self, key): return self._data[key] def __setitem__(self, key, value): # 如果该键有定义验证规则 if key in self.VALIDATORS: validator, converter = self.VALIDATORS[key] # 首先尝试转换（如果提供了转换函数） if converter is not None: try: value = converter(value) except (ValueError, TypeError) as e: raise ValueError(f"无法将键 {key!r} 的值 {value!r} 转换为合适类型: {e}") # 然后验证 if not validator(value): raise ValueError(f"键 {key!r} 的值 {value!r} 不符合验证规则") # 存储验证/转换后的值 self._data[key] = value def __repr__(self): return f"ValidatedConfigDict({self._data})" # 使用示例 config = ValidatedConfigDict() config['port'] = 8080 # 正常 print(config['port']) # 8080 try: config['port'] = 80 # 端口太小 except ValueError as e: print(f"错误: {e}") # 输出: 错误: 键 'port' 的值 80 不符合验证规则 config['debug'] = 'true' # 字符串会自动转换为布尔值 print(config['debug']) # True config['log_level'] = 'info' # 自动转换为大写 print(config['log_level']) # INFO # 未在VALIDATORS中定义的键可以自由存储，不受限制 config['custom_key'] = 'any value' print(config['custom_key']) # any value ``` 这个实现有几个值得注意的地方： - **分离验证与转换**：我们将验证逻辑和转换逻辑分开定义。转换函数负责将输入值标准化（如字符串转布尔、小写转大写），验证函数则检查标准化后的值是否有效。 - **灵活的验证规则**：验证规则使用可调用对象（函数或lambda表达式），这意味着你可以实现非常复杂的验证逻辑，比如检查网络地址格式、文件路径是否存在等。 - **优雅的降级**：对于未在`VALIDATORS`中定义的键，我们不做任何处理，直接存储。这使得字典既能保证关键配置项的正确性，又保持了灵活性。 > **提示**：在实际项目中，你可能会将验证规则定义在类外部（如JSON或YAML文件中），以便在不修改代码的情况下更新验证规则。`__setitem__`方法可以从配置文件动态加载验证逻辑。 除了验证，另一个常见需求是**自动类型转换**。比如，我们希望所有数值都以`decimal.Decimal`对象的形式存储，以确保精确计算： ```python from decimal import Decimal, InvalidOperation class AutoDecimalDict: """自动将数值转换为Decimal的字典""" def __init__(self): self._data = {} def __getitem__(self, key): return self._data[key] def __setitem__(self, key, value): # 尝试将值转换为Decimal if isinstance(value, (int, float, str)): try: value = Decimal(str(value)) except InvalidOperation: pass # 转换失败，保持原值 # 如果是Decimal类型，直接存储 elif isinstance(value, Decimal): pass # 其他类型保持不变 self._data[key] = value def __repr__(self): return f"AutoDecimalDict({self._data})" # 使用 price_dict = AutoDecimalDict() price_dict['apple'] = 3.99 # 浮点数会被转换为Decimal price_dict['banana'] = '2.50' # 字符串也会被转换 price_dict['description'] = 'Fresh fruit' # 非数值保持不变 print(price_dict['apple']) # Decimal('3.99') print(price_dict['apple'] * 2) # Decimal('7.98')，精确计算 ``` 这种自动转换在财务计算、科学计算等对精度要求高的场景中非常有用。通过在`__setitem__`中统一处理，我们确保了整个字典中数值类型的一致性。 ## 3. 高级模式：实现惰性加载与缓存集成 当字典需要处理大量数据或访问远程资源时，每次`__getitem__`都直接返回数据可能效率低下。这时，我们可以实现**惰性加载**（Lazy Loading）：只有在第一次访问某个键时才真正加载数据，并将结果缓存起来供后续使用。结合`__getitem__`和`__setitem__`，我们可以创建出智能的缓存字典。 ### 3.1 基础惰性加载字典 首先，我们实现一个简单的惰性加载字典，它接受一个生成器函数，该函数知道如何根据键加载数据： ```python class LazyLoadingDict: """惰性加载字典，只在第一次访问时加载数据""" def __init__(self, loader_func): """ Args: loader_func: 一个函数，接受键作为参数，返回对应的值 """ self._loader = loader_func self._cache = {} # 缓存已加载的数据 self._loaded_keys = set() # 记录哪些键已经加载过 def __getitem__(self, key): # 如果键不在缓存中，使用loader函数加载 if key not in self._loaded_keys: try: self._cache[key] = self._loader(key) self._loaded_keys.add(key) except Exception as e: # 加载失败，可以抛出KeyError或其他异常 raise KeyError(f"无法加载键 {key!r}: {e}") return self._cache[key] def __setitem__(self, key, value): # 直接设置值会覆盖缓存 self._cache[key] = value self._loaded_keys.add(key) def __contains__(self, key): # 注意：我们不知道loader能否加载这个键，所以只检查已加载的键 return key in self._loaded_keys def preload(self, *keys): """预加载一个或多个键""" for key in keys: _ = self[key] # 访问键会触发加载 def clear_cache(self): """清空缓存，但保留loader函数""" self._cache.clear() self._loaded_keys.clear() def __repr__(self): loaded = {k: v for k, v in self._cache.items() if k in self._loaded_keys} return f"LazyLoadingDict(loaded={loaded})" # 示例：模拟从数据库加载用户信息 def load_user_from_db(user_id): """模拟从数据库加载用户信息的函数（实际中可能涉及网络IO）""" print(f"[DB] 正在从数据库加载用户 {user_id}...") # 模拟耗时操作 import time time.sleep(0.5) # 模拟返回数据 users = { 1: {'name': 'Alice', 'email': 'alice@example.com'}, 2: {'name': 'Bob', 'email': 'bob@example.com'}, 3: {'name': 'Charlie', 'email': 'charlie@example.com'}, } return users.get(user_id, {'name': 'Unknown', 'email': ''}) # 使用 user_cache = LazyLoadingDict(load_user_from_db) print("第一次访问用户1:") print(user_cache[1]) # 会输出"[DB] 正在从数据库加载用户 1..."，然后返回数据 print("\n第二次访问用户1（从缓存获取）:") print(user_cache[1]) # 直接从缓存获取，不会再次访问数据库 # 预加载多个用户 print("\n预加载用户2和3:") user_cache.preload(2, 3) print(user_cache[2]) # 从缓存获取，无数据库访问 ``` 这个实现的关键点在于： - `__getitem__`首先检查键是否在`_loaded_keys`集合中。如果不在，说明还没有加载过，于是调用`loader_func`加载数据，并将结果存入`_cache`，同时标记该键为已加载。 - 一旦数据被加载，后续访问就直接从缓存返回，避免了重复的加载操作。 - `__setitem__`允许手动设置值，这会直接更新缓存，适用于需要手动更新数据的场景。 ### 3.2 集成外部缓存系统（如Redis） 在分布式系统中，我们经常需要将数据缓存到Redis等外部存储中，以便多个进程或服务器共享缓存。我们可以创建一个字典类，它自动将数据同步到Redis，同时在本地内存中维护一个快速缓存（两级缓存）。 > **注意**：以下示例假设你已经安装了`redis` Python包，并且有一个Redis服务器在运行。实际使用时需要处理连接错误、序列化等问题。 ```python import json import pickle from typing import Any, Optional class RedisBackedDict: """使用Redis作为后端存储的字典，支持本地内存缓存""" def __init__(self, redis_client, namespace: str = "cache", local_cache_size: int = 1000, serializer: str = "json"): """ Args: redis_client: redis.Redis实例 namespace: 用于在Redis中区分不同字典的命名空间 local_cache_size: 本地LRU缓存的大小（0表示禁用本地缓存） serializer: 序列化方式，'json'或'pickle' """ self.redis = redis_client self.namespace = namespace self.serializer = serializer # 设置本地缓存（使用有序字典模拟LRU） self._local_cache = {} self._local_cache_order = [] self._local_cache_maxsize = local_cache_size # 用于生成Redis键的前缀 self._key_prefix = f"{namespace}:" def _make_redis_key(self, key: Any) -> str: """将任意键转换为Redis中使用的字符串键""" # 对于简单类型，直接转换为字符串 if isinstance(key, (str, int, float)): return f"{self._key_prefix}{key}" # 对于复杂类型，使用序列化 if self.serializer == "json": key_str = json.dumps(key, sort_keys=True) else: # pickle key_str = pickle.dumps(key).hex() return f"{self._key_prefix}{key_str}" def _serialize(self, value: Any) -> str: """序列化值以便存储到Redis""" if self.serializer == "json": return json.dumps(value) else: # pickle return pickle.dumps(value) def _deserialize(self, data: str) -> Any: """从Redis中反序列化值""" if self.serializer == "json": return json.loads(data) else: # pickle return pickle.loads(data) def __getitem__(self, key): redis_key = self._make_redis_key(key) # 首先检查本地缓存 if redis_key in self._local_cache: # 更新LRU顺序（将最近访问的键移到末尾） self._local_cache_order.remove(redis_key) self._local_cache_order.append(redis_key) return self._local_cache[redis_key] # 本地缓存未命中，从Redis获取 data = self.redis.get(redis_key) if data is None: raise KeyError(f"键 {key!r} 不存在于Redis中") # 反序列化 value = self._deserialize(data) # 存入本地缓存 self._update_local_cache(redis_key, value) return value def __setitem__(self, key, value): redis_key = self._make_redis_key(key) # 序列化并存储到Redis serialized = self._serialize(value) self.redis.set(redis_key, serialized) # 更新本地缓存 self._update_local_cache(redis_key, value) def _update_local_cache(self, redis_key: str, value: Any): """更新本地LRU缓存""" if self._local_cache_maxsize <= 0: return if redis_key in self._local_cache: # 键已存在，更新值并调整顺序 self._local_cache[redis_key] = value self._local_cache_order.remove(redis_key) self._local_cache_order.append(redis_key) else: # 键不存在，添加新条目 if len(self._local_cache) >= self._local_cache_maxsize: # 缓存已满，移除最久未使用的键 lru_key = self._local_cache_order.pop(0) del self._local_cache[lru_key] self._local_cache[redis_key] = value self._local_cache_order.append(redis_key) def __delitem__(self, key): redis_key = self._make_redis_key(key) # 从Redis删除 if not self.redis.delete(redis_key): raise KeyError(f"键 {key!r} 不存在于Redis中") # 从本地缓存删除 if redis_key in self._local_cache: del self._local_cache[redis_key] self._local_cache_order.remove(redis_key) def __contains__(self, key): redis_key = self._make_redis_key(key) # 先检查本地缓存 if redis_key in self._local_cache: return True # 本地缓存未命中，检查Redis return self.redis.exists(redis_key) == 1 def clear(self): """清空整个字典（包括Redis中的所有相关键）""" # 查找所有以_prefix开头的键 keys = self.redis.keys(f"{self._key_prefix}*") if keys: self.redis.delete(*keys) # 清空本地缓存 self._local_cache.clear() self._local_cache_order.clear() def get(self, key, default=None): """安全的获取方法，键不存在时返回默认值""" try: return self[key] except KeyError: return default # 使用示例（需要实际的Redis连接） try: import redis # 创建Redis连接（实际使用时需要配置正确的host、port等参数） r = redis.Redis(host='localhost', port=6379, db=0, decode_responses=False) # 测试连接 r.ping() # 创建Redis支持的字典 cache_dict = RedisBackedDict(r, namespace="myapp", local_cache_size=100) # 存储数据（会自动同步到Redis） cache_dict['user:1001'] = {'name': 'Alice', 'score': 95} cache_dict['config:timeout'] = 30 # 获取数据（优先从本地缓存，其次从Redis） print(cache_dict['user:1001']) # {'name': 'Alice', 'score': 95} # 即使重启Python进程，数据仍然在Redis中 print("数据已持久化到Redis") except ImportError: print("请先安装redis包: pip install redis") except redis.ConnectionError: print("无法连接到Redis服务器，请确保Redis正在运行") ``` 这个`RedisBackedDict`类展示了如何将`__getitem__`和`__setitem__`与外部系统集成： 1. **两级缓存架构**：本地内存缓存（LRU策略）提供快速访问，Redis提供持久化和进程间共享。 2. **键的命名空间**：使用`namespace`前缀避免不同字典之间的键冲突。 3. **灵活的序列化**：支持JSON和Pickle两种序列化方式。JSON可读性好且跨语言，Pickle支持更多Python类型但可能存在安全风险。 4. **透明的数据同步**：用户只需像使用普通字典一样操作，所有与Redis的交互都在幕后自动完成。 在实际生产环境中，你可能还需要考虑： - **连接池管理**：重用Redis连接以提高性能。 - **异常处理**：网络故障时的重试机制和降级策略。 - **过期时间**：为键设置TTL（Time-To-Live），让数据自动过期。 - **事务支持**：确保多个操作的原子性。 ## 4. 创新实践：构建具有动态计算能力的字典 有时候，我们需要的不仅仅是存储静态数据，而是希望字典能够根据键**动态计算**值。这种模式在实现计算属性、延迟计算或派生数据时非常有用。通过结合`__getitem__`和描述符（Descriptor）或属性（Property），我们可以创建出真正“智能”的字典。 ### 4.1 计算属性字典 假设我们有一个表示向量的字典，存储了`x`、`y`、`z`分量。我们希望能够通过键`'magnitude'`直接获取向量的模长，而不需要显式计算： ```python import math class VectorDict: """向量字典，支持动态计算属性""" def __init__(self, x=0, y=0, z=0): self._components = {'x': float(x), 'y': float(y), 'z': float(z)} # 定义计算属性 self._computed = { 'magnitude': lambda: math.sqrt( self._components['x']**2 + self._components['y']**2 + self._components['z']**2 ), 'unit_vector': lambda: { 'x': self._components['x'] / self['magnitude'] if self['magnitude'] != 0 else 0, 'y': self._components['y'] / self['magnitude'] if self['magnitude'] != 0 else 0, 'z': self._components['z'] / self['magnitude'] if self['magnitude'] != 0 else 0, }, 'xy_angle': lambda: math.atan2(self._components['y'], self._components['x']), } def __getitem__(self, key): # 首先检查是否是计算属性 if key in self._computed: return self._computed[key]() # 然后检查是否是基础分量 if key in self._components: return self._components[key] # 最后检查是否有默认值 if hasattr(self, f'_default_{key}'): return getattr(self, f'_default_{key}')() raise KeyError(f"键 {key!r} 不存在") def __setitem__(self, key, value): # 计算属性是只读的 if key in self._computed: raise KeyError(f"计算属性 {key!r} 是只读的") # 只能设置基础分量 if key in self._components: self._components[key] = float(value) else: # 允许动态添加新的基础属性 self._components[key] = value def __contains__(self, key): return key in self._components or key in self._computed def add_computed(self, name, compute_func): """动态添加计算属性""" self._computed[name] = compute_func def __repr__(self): return f"VectorDict(x={self._components['x']}, y={self._components['y']}, z={self._components['z']})" # 使用示例 v = VectorDict(3, 4, 0) print(f"向量: {v}") print(f"x分量: {v['x']}") # 3.0 print(f"模长: {v['magnitude']}") # 5.0 print(f"单位向量: {v['unit_vector']}") # {'x': 0.6, 'y': 0.8, 'z': 0.0} print(f"XY平面角度: {v['xy_angle']}") # 约0.93弧度 # 尝试修改计算属性会报错 try: v['magnitude'] = 10 except KeyError as e: print(f"错误: {e}") # 错误: 计算属性 'magnitude' 是只读的 # 动态添加计算属性 v.add_computed('xy_magnitude', lambda: math.sqrt(v['x']**2 + v['y']**2)) print(f"XY平面模长: {v['xy_magnitude']}") # 5.0 ``` 这个实现的核心思想是： - 将数据分为两类：**基础数据**（存储在`_components`中）和**计算属性**（存储在`_computed`中，值为计算函数）。 - `__getitem__`首先检查键是否在`_computed`中，如果是，就调用对应的计算函数并返回结果。这使得每次访问计算属性时都能得到最新的计算结果。 - 计算属性是只读的，尝试通过`__setitem__`修改它们会抛出异常。 - 支持动态添加新的计算属性，这为扩展字典功能提供了极大灵活性。 ### 4.2 响应式字典：自动更新依赖项 在更复杂的场景中，一个计算属性可能依赖于其他属性，当依赖项变化时，我们希望计算属性能自动更新（或标记为需要重新计算）。这需要引入**响应式**（Reactive）机制： ```python class ReactiveDict: """响应式字典，自动管理计算属性的依赖关系""" def __init__(self): self._data = {} self._computed = {} # 计算属性定义 self._dependencies = {} # 计算属性依赖哪些数据键 self._reverse_deps = {} # 每个数据键被哪些计算属性依赖 self._cache = {} # 计算结果的缓存 self._cache_valid = {} # 缓存是否有效 def define_computed(self, name, compute_func, deps): """ 定义计算属性 Args: name: 属性名 compute_func: 计算函数 deps: 依赖的数据键列表 """ self._computed[name] = compute_func self._dependencies[name] = set(deps) # 建立反向依赖关系 for dep in deps: if dep not in self._reverse_deps: self._reverse_deps[dep] = set() self._reverse_deps[dep].add(name) # 初始时缓存无效 self._cache_valid[name] = False def __getitem__(self, key): # 如果是计算属性 if key in self._computed: # 如果缓存无效或不存在，重新计算 if not self._cache_valid.get(key, False): # 收集依赖项的值 dep_values = {} for dep in self._dependencies[key]: dep_values[dep] = self[dep] # 递归获取 # 计算并缓存结果 self._cache[key] = self._computed[key](**dep_values) self._cache_valid[key] = True return self._cache[key] # 如果是普通数据 if key in self._data: return self._data[key] raise KeyError(f"键 {key!r} 不存在") def __setitem__(self, key, value): # 设置数据 self._data[key] = value # 如果这个键被某些计算属性依赖，使那些缓存失效 if key in self._reverse_deps: for computed_key in self._reverse_deps[key]: self._cache_valid[computed_key] = False def __contains__(self, key): return key in self._data or key in self._computed def invalidate(self, key=None): """使缓存失效""" if key is None: # 使所有缓存失效 for k in self._cache_valid: self._cache_valid[k] = False elif key in self._cache_valid: # 使指定键的缓存失效 self._cache_valid[key] = False # 使用示例：购物车总价计算 cart = ReactiveDict() # 设置商品价格和数量 cart['apple_price'] = 3.5 cart['apple_qty'] = 2 cart['banana_price'] = 2.0 cart['banana_qty'] = 3 # 定义计算属性 cart.define_computed( 'apple_total', lambda apple_price, apple_qty: apple_price * apple_qty, deps=['apple_price', 'apple_qty'] ) cart.define_computed( 'banana_total', lambda banana_price, banana_qty: banana_price * banana_qty, deps=['banana_price', 'banana_qty'] ) cart.define_computed( 'grand_total', lambda apple_total, banana_total: apple_total + banana_total, deps=['apple_total', 'banana_total'] ) print(f"苹果总价: {cart['apple_total']}") # 7.0 print(f"香蕉总价: {cart['banana_total']}") # 6.0 print(f"购物车总价: {cart['grand_total']}") # 13.0 # 修改苹果数量，所有相关计算属性会自动更新 print("\n修改苹果数量为3...") cart['apple_qty'] = 3 # 再次访问计算属性，会触发重新计算 print(f"苹果总价: {cart['apple_total']}") # 10.5 print(f"购物车总价: {cart['grand_total']}") # 16.5 # 查看内部状态 print(f"\n缓存状态: {cart._cache_valid}") ``` 这个`ReactiveDict`实现了简单的响应式系统： 1. **依赖跟踪**：当定义计算属性时，我们明确指定它依赖哪些数据键。系统会记录这些依赖关系。 2. **缓存管理**：计算属性的结果被缓存起来。当依赖的数据发生变化时，相关的计算属性缓存被标记为无效。 3. **惰性重新计算**：只有当访问一个缓存无效的计算属性时，才会真正重新计算。这避免了不必要的计算开销。 这种模式在需要复杂派生数据的应用中非常有用，比如电子表格、仪表板、实时数据分析等场景。你可以进一步扩展它，添加： - **脏检查**：定期或在特定时机检查哪些数据发生了变化。 - **批量更新**：在一次事务中更新多个数据键，然后统一使缓存失效。 - **异步计算**：对于耗时的计算，使用异步函数并在后台更新缓存。 通过`__getitem__`和`__setitem__`，我们不仅能够存储和检索数据，还能嵌入复杂的业务逻辑，创建出真正智能的数据容器。这种能力是Python动态性和元编程魅力的完美体现。