描述符协议
数据验证描述符
class ValidString:
"""字符串验证描述符"""
def __init__(self, minlen=0, maxlen=None):
self.minlen = minlen
self.maxlen = maxlen
def __get__(self, instance, owner):
if instance is None:
return self
return instance.__dict__.get(self.name)
def __set__(self, instance, value):
if not isinstance(value, str):
raise TypeError("值必须是字符串类型")
if len(value) < self.minlen:
raise ValueError(f"字符串长度不能小于{self.minlen}")
if self.maxlen and len(value) > self.maxlen:
raise ValueError(f"字符串长度不能大于{self.maxlen}")
instance.__dict__[self.name] = value
def __set_name__(self, owner, name):
self.name = name
class User:
name = ValidString(minlen=2, maxlen=20)
email = ValidString(minlen=5, maxlen=50)
def __init__(self, name, email):
self.name = name
self.email = email
# 使用示例
user = User("张三", "[email protected]")
# user.name = "A" # 会引发ValueError
延迟计算描述符
class LazyProperty:
"""延迟计算属性描述符"""
def __init__(self, func):
self.func = func
self.name = func.__name__
def __get__(self, instance, owner):
if instance is None:
return self
# 计算值并缓存
value = self.func(instance)
setattr(instance, self.name, value)
return value
class Circle:
def __init__(self, radius):
self.radius = radius
@LazyProperty
def area(self):
print("计算面积...")
import math
return math.pi * self.radius ** 2
@LazyProperty
def perimeter(self):
print("计算周长...")
import math
return 2 * math.pi * self.radius
# 使用示例
circle = Circle(5)
print(circle.area) # 第一次调用会计算
print(circle.area) # 第二次调用直接返回缓存值
上下文管理
自定义上下文管理器
class Timer:
"""计时器上下文管理器"""
def __init__(self, name="Timer"):
self.name = name
def __enter__(self):
import time
self.start_time = time.time()
return self
def __exit__(self, exc_type, exc_val, exc_tb):
import time
self.end_time = time.time()
self.duration = self.end_time - self.start_time
print(f"{self.name} 耗时: {self.duration:.2f}秒")
# 使用示例
def process_data():
import time
time.sleep(1) # 模拟耗时操作
with Timer("数据处理"):
process_data()
数据库连接管理器
class DatabaseConnection:
"""数据库连接管理器"""
def __init__(self, host, port, database):
self.host = host
self.port = port
self.database = database
self.connection = None
def __enter__(self):
print(f"连接到数据库 {self.host}:{self.port}/{self.database}")
self.connection = self._connect()
return self.connection
def __exit__(self, exc_type, exc_val, exc_tb):
if self.connection:
print("关闭数据库连接")
self.connection.close()
if exc_type:
print(f"发生错误: {exc_val}")
return False # 重新抛出异常
def _connect(self):
# 模拟数据库连接
class Connection:
def execute(self, sql):
print(f"执行SQL: {sql}")
def close(self):
pass
return Connection()
# 使用示例
with DatabaseConnection("localhost", 5432, "mydb") as conn:
conn.execute("SELECT * FROM users")
高级设计模式
工厂方法模式
from abc import ABC, abstractmethod
class Document(ABC):
"""文档抽象类"""
@abstractmethod
def create(self):
pass
class PDFDocument(Document):
def create(self):
return "创建PDF文档"
class WordDocument(Document):
def create(self):
return "创建Word文档"
class DocumentFactory:
"""文档工厂"""
_creators = {}
@classmethod
def register_format(cls, format_type, creator):
"""注册文档创建器"""
cls._creators[format_type.lower()] = creator
@classmethod
def create_document(cls, format_type):
"""创建文档"""
creator = cls._creators.get(format_type.lower())
if not creator:
raise ValueError(f"不支持的文档格式: {format_type}")
return creator()
# 注册文档类型
DocumentFactory.register_format("PDF", PDFDocument)
DocumentFactory.register_format("WORD", WordDocument)
# 使用示例
doc = DocumentFactory.create_document("PDF")
print(doc.create())
状态模式
from abc import ABC, abstractmethod
class State(ABC):
"""状态抽象类"""
@abstractmethod
def handle(self, context):
pass
class Context:
"""上下文类"""
def __init__(self, state):
self._state = state
def set_state(self, state):
self._state = state
def request(self):
self._state.handle(self)
class ConcreteStateA(State):
def handle(self, context):
print("处理状态A的请求")
context.set_state(ConcreteStateB())
class ConcreteStateB(State):
def handle(self, context):
print("处理状态B的请求")
context.set_state(ConcreteStateA())
# 使用示例
context = Context(ConcreteStateA())
context.request() # 输出：处理状态A的请求
context.request() # 输出：处理状态B的请求
实际应用
配置管理系统
class Config:
"""配置管理类"""
_instance = None
_initialized = False
def __new__(cls):
if cls._instance is None:
cls._instance = super().__new__(cls)
return cls._instance
def __init__(self):
if not self._initialized:
self._config = {}
self._load_config()
self._initialized = True
def _load_config(self):
"""加载配置"""
# 模拟从文件加载配置
self._config = {
'debug': True,
'database': {
'host': 'localhost',
'port': 5432
}
}
def get(self, key, default=None):
"""获取配置"""
return self._config.get(key, default)
def set(self, key, value):
"""设置配置"""
self._config[key] = value
self._save_config()
def _save_config(self):
"""保存配置"""
print("保存配置到文件")
# 使用示例
config = Config()
print(config.get('database'))
config2 = Config() # 返回同一个实例
print(config is config2) # 输出：True
事件系统
class Event:
"""事件类"""
def __init__(self, event_type, data=None):
self.type = event_type
self.data = data
self.timestamp = time.time()
class EventDispatcher:
"""事件调度器"""
def __init__(self):
self._handlers = {}
def register_handler(self, event_type, handler):
"""注册事件处理器"""
if event_type not in self._handlers:
self._handlers[event_type] = []
self._handlers[event_type].append(handler)
def dispatch_event(self, event):
"""分发事件"""
if event.type in self._handlers:
for handler in self._handlers[event.type]:
handler(event)
class Application:
"""应用程序类"""
def __init__(self):
self.dispatcher = EventDispatcher()
self._setup_handlers()
def _setup_handlers(self):
"""设置事件处理器"""
self.dispatcher.register_handler(
"user_login",
self._handle_user_login
)
self.dispatcher.register_handler(
"user_logout",
self._handle_user_logout
)
def _handle_user_login(self, event):
print(f"用户登录: {event.data}")
def _handle_user_logout(self, event):
print(f"用户登出: {event.data}")
def login(self, username):
"""用户登录"""
event = Event("user_login", username)
self.dispatcher.dispatch_event(event)
def logout(self, username):
"""用户登出"""
event = Event("user_logout", username)
self.dispatcher.dispatch_event(event)
# 使用示例
app = Application()
app.login("张三")
app.logout("张三")