描述符协议实战详解
【摘要】 描述符协议实战详解描述符(Descriptor)是Python中实现属性访问控制的核心机制,它支撑着@property、staticmethod、classmethod等特性的实现。本文将深入解析描述符协议及其实战应用。 描述符基础概念描述符是实现了以下任一方法的类:__get__(self, obj, type=None):获取属性值__set__(self, obj, value):设...
描述符协议实战详解
描述符(Descriptor)是Python中实现属性访问控制的核心机制,它支撑着@property、staticmethod、classmethod等特性的实现。本文将深入解析描述符协议及其实战应用。
描述符基础概念
描述符是实现了以下任一方法的类:
__get__(self, obj, type=None):获取属性值__set__(self, obj, value):设置属性值__delete__(self, obj):删除属性
根据实现的方法,描述符分为:
- 非数据描述符:仅实现
__get__ - 数据描述符:实现
__set__或__delete__
描述符核心实现
"""
Python描述符协议实战详解
包含属性验证、懒加载、类型检查等高级应用
"""
import weakref
from typing import Any, Callable, Optional, Type, Union
from functools import wraps
import time
# ============ 基础描述符示例 ============
class Validator:
"""基础验证描述符"""
def __init__(self, min_value=None, max_value=None):
self.min_value = min_value
self.max_value = max_value
self.name = None
self.values = weakref.WeakKeyDictionary()
def __set_name__(self, owner, name):
"""Python 3.6+ 自动设置属性名"""
self.name = name
def __get__(self, obj, objtype=None):
if obj is None:
return self
return self.values.get(obj, None)
def __set__(self, obj, value):
self.validate(value)
self.values[obj] = value
def validate(self, value):
"""子类重写验证逻辑"""
pass
class Integer(Validator):
"""整数类型描述符"""
def validate(self, value):
if not isinstance(value, int):
raise TypeError(f"{self.name} 必须是整数类型")
if self.min_value is not None and value < self.min_value:
raise ValueError(f"{self.name} 必须 >= {self.min_value}")
if self.max_value is not None and value > self.max_value:
raise ValueError(f"{self.name} 必须 <= {self.max_value}")
class String(Validator):
"""字符串类型描述符"""
def __init__(self, min_length=None, max_length=None, pattern=None):
super().__init__()
self.min_length = min_length
self.max_length = max_length
self.pattern = pattern
def validate(self, value):
if not isinstance(value, str):
raise TypeError(f"{self.name} 必须是字符串类型")
if self.min_length is not None and len(value) < self.min_length:
raise ValueError(f"{self.name} 长度必须 >= {self.min_length}")
if self.max_length is not None and len(value) > self.max_length:
raise ValueError(f"{self.name} 长度必须 <= {self.max_length}")
class Float(Validator):
"""浮点数类型描述符"""
def validate(self, value):
if not isinstance(value, (int, float)):
raise TypeError(f"{self.name} 必须是数值类型")
value = float(value)
if self.min_value is not None and value < self.min_value:
raise ValueError(f"{self.name} 必须 >= {self.min_value}")
if self.max_value is not None and value > self.max_value:
raise ValueError(f"{self.name} 必须 <= {self.max_value}")
# ============ 懒加载描述符 ============
class LazyProperty:
"""懒加载属性描述符"""
def __init__(self, func):
self.func = func
self.name = func.__name__
self.values = weakref.WeakKeyDictionary()
def __get__(self, obj, objtype=None):
if obj is None:
return self
if obj not in self.values:
# 首次访问时计算并缓存
start_time = time.time()
value = self.func(obj)
self.values[obj] = value
elapsed = time.time() - start_time
print(f"[Lazy] {self.name} 计算完成,耗时: {elapsed:.4f}s")
return self.values[obj]
def __set__(self, obj, value):
raise AttributeError(f"{self.name} 是只读属性")
# ============ 类型检查描述符 ============
class TypedAttribute:
"""类型检查属性描述符"""
def __init__(self, expected_type: Type):
self.expected_type = expected_type
self.name = None
self.values = weakref.WeakKeyDictionary()
def __set_name__(self, owner, name):
self.name = name
def __get__(self, obj, objtype=None):
if obj is None:
return self
return self.values.get(obj)
def __set__(self, obj, value):
if not isinstance(value, self.expected_type):
raise TypeError(
f"{self.name} 期望类型 {self.expected_type.__name__}, "
f"实际类型 {type(value).__name__}"
)
self.values[obj] = value
# ============ 审计日志描述符 ============
class AuditedAttribute:
"""带审计日志的属性描述符"""
def __init__(self, sensitive=False):
self.sensitive = sensitive
self.name = None
self.values = weakref.WeakKeyDictionary()
self.audit_log = []
def __set_name__(self, owner, name):
self.name = name
def __get__(self, obj, objtype=None):
if obj is None:
return self
value = self.values.get(obj)
self._log_access("READ", obj, value)
return value
def __set__(self, obj, value):
old_value = self.values.get(obj)
self.values[obj] = value
self._log_access("WRITE", obj, value, old_value)
def __delete__(self, obj):
old_value = self.values.get(obj)
del self.values[obj]
self._log_access("DELETE", obj, None, old_value)
def _log_access(self, operation, obj, value, old_value=None):
"""记录访问日志"""
display_value = "***" if self.sensitive and value else value
display_old = "***" if self.sensitive and old_value else old_value
log_entry = {
"timestamp": time.strftime("%Y-%m-%d %H:%M:%S"),
"attribute": self.name,
"operation": operation,
"new_value": display_value,
"old_value": display_old,
"object_id": id(obj)
}
self.audit_log.append(log_entry)
print(f"[Audit] {operation}: {self.name} = {display_value}")
# ============ 缓存描述符 ============
class CachedProperty:
"""带过期时间的缓存属性"""
def __init__(self, ttl_seconds: float = 60.0):
self.ttl_seconds = ttl_seconds
self.func = None
self.name = None
self.cache = weakref.WeakKeyDictionary()
self.timestamps = weakref.WeakKeyDictionary()
def __call__(self, func):
self.func = func
self.name = func.__name__
return self
def __get__(self, obj, objtype=None):
if obj is None:
return self
now = time.time()
cached_time = self.timestamps.get(obj, 0)
# 检查缓存是否过期
if obj not in self.cache or (now - cached_time) > self.ttl_seconds:
self.cache[obj] = self.func(obj)
self.timestamps[obj] = now
print(f"[Cache] {self.name} 已更新缓存")
else:
remaining = self.ttl_seconds - (now - cached_time)
print(f"[Cache] {self.name} 使用缓存 (剩余{remaining:.1f}s)")
return self.cache[obj]
def invalidate(self, obj):
"""手动使缓存失效"""
if obj in self.cache:
del self.cache[obj]
del self.timestamps[obj]
# ============ 使用描述符的模型类 ============
class User:
"""使用描述符的用户类"""
# 类型验证描述符
id = Integer(min_value=1)
username = String(min_length=3, max_length=20)
email = String(min_length=5, max_length=100)
age = Integer(min_value=0, max_value=150)
score = Float(min_value=0.0, max_value=100.0)
# 审计日志描述符
password = AuditedAttribute(sensitive=True)
# 类型检查描述符
metadata = TypedAttribute(dict)
def __init__(self, id, username, email, age=0):
self.id = id
self.username = username
self.email = email
self.age = age
self.score = 0.0
self.metadata = {}
@LazyProperty
def profile_completeness(self):
"""计算资料完整度(懒加载)"""
time.sleep(0.1) # 模拟耗时计算
fields = [self.username, self.email, self.age]
filled = sum(1 for f in fields if f)
return (filled / len(fields)) * 100
@CachedProperty(ttl_seconds=5.0)
def reputation_score(self):
"""计算信誉分数(带缓存)"""
time.sleep(0.05)
base_score = self.score * 10
age_bonus = min(self.age * 0.5, 20)
return base_score + age_bonus
class Product:
"""产品类 - 展示更多描述符应用"""
name = String(min_length=1, max_length=100)
price = Float(min_value=0.01)
stock = Integer(min_value=0)
def __init__(self, name, price, stock=0):
self.name = name
self.price = price
self.stock = stock
@property
def is_available(self):
"""是否可购买"""
return self.stock > 0
@LazyProperty
def total_value(self):
"""库存总价值"""
return self.price * self.stock
def main():
"""主函数 - 演示描述符应用"""
print("="*60)
print("Python描述符协议实战演示")
print("="*60)
# 1. 类型验证演示
print("\n【类型验证演示】")
user = User(id=1, username="alice", email="alice@example.com", age=25)
print(f"创建用户: {user.username}, 年龄: {user.age}")
# 尝试设置无效值
try:
user.age = -5
except ValueError as e:
print(f"验证错误: {e}")
try:
user.username = "ab" # 太短
except ValueError as e:
print(f"验证错误: {e}")
# 2. 懒加载演示
print("\n【懒加载演示】")
print("首次访问 profile_completeness:")
completeness = user.profile_completeness
print(f"资料完整度: {completeness:.1f}%")
print("\n再次访问 profile_completeness(使用缓存):")
completeness = user.profile_completeness
print(f"资料完整度: {completeness:.1f}%")
# 3. 审计日志演示
print("\n【审计日志演示】")
user.password = "secret123"
_ = user.password # 读取
# 4. 缓存属性演示
print("\n【缓存属性演示】")
user.score = 8.5
print(f"信誉分数: {user.reputation_score:.2f}")
print(f"信誉分数(缓存): {user.reputation_score:.2f}")
time.sleep(6) # 等待缓存过期
print(f"信誉分数(过期后重新计算): {user.reputation_score:.2f}")
# 5. 产品类演示
print("\n【产品类演示】")
product = Product(name="Laptop", price=999.99, stock=50)
print(f"产品: {product.name}, 价格: ${product.price}")
print(f"库存价值: ${product.total_value:.2f}")
print(f"是否可购买: {product.is_available}")
# 6. 类型检查演示
print("\n【类型检查演示】")
user.metadata = {"role": "admin", "department": "IT"}
print(f"用户元数据: {user.metadata}")
try:
user.metadata = "invalid" # 类型错误
except TypeError as e:
print(f"类型错误: {e}")
print("\n" + "="*60)
print("描述符应用总结")
print("="*60)
print("1. 数据验证: 自动验证属性值的类型和范围")
print("2. 懒加载: 延迟计算属性值,提高启动性能")
print("3. 审计日志: 记录属性的访问和修改历史")
print("4. 缓存机制: 缓存计算结果,减少重复计算")
print("5. 类型检查: 运行时类型验证")
print("="*60)
if __name__ == "__main__":
main()
描述符查找流程图
描述符优先级规则
关键要点
- 数据描述符优先:同时实现
__get__和__set__的描述符优先级最高 - 实例属性次之:如果属性存在于实例的
__dict__中,会覆盖非数据描述符 - 内存管理:使用
weakref.WeakKeyDictionary避免循环引用 - Python 3.6+特性:
__set_name__自动获取属性名 - 应用场景:ORM字段、属性验证、懒加载、缓存、审计日志
描述符是Python属性系统的基石,掌握描述符协议可以编写出更加优雅、高效的代码。
【声明】本内容来自华为云开发者社区博主,不代表华为云及华为云开发者社区的观点和立场。转载时必须标注文章的来源(华为云社区)、文章链接、文章作者等基本信息,否则作者和本社区有权追究责任。如果您发现本社区中有涉嫌抄袭的内容,欢迎发送邮件进行举报,并提供相关证据,一经查实,本社区将立刻删除涉嫌侵权内容,举报邮箱:
cloudbbs@huaweicloud.com
- 点赞
- 收藏
- 关注作者
评论(0)