python

# Python高级特性技能 ## 概述 Python提供了丰富的高级特性，包括装饰器、生成器、元类、异步编程等，这些特性能够显著提升代码的表达力和性能。不当的高级特性使用会导致代码难以理解、性能下降、维护困难。 **核心原则**: 好的Python高级代码应该优雅简洁、性能优良、可读性强、易于维护。坏的高级代码会过度抽象、性能损耗、难以调试。 ## 何时使用 **始终:** - 构建复杂系统时 - 优化代码性能时 - 处理并发编程时 - 实现框架设计时 - 提升代码复用时 - 解决技术难题时 **触发短语:** - "Python装饰器怎么用？" - "生成器和迭代器区别" - "Python异步编程最佳实践" - "元类应用场景" - "Python性能优化技巧" - "高级数据结构应用" ## Python高级特性技能功能 ### 函数式编程 - 装饰器模式 - 生成器和迭代器 - 闭包和高阶函数 - lambda表达式 - 函数式工具 ### 面向对象高级 - 元类编程 - 描述符协议 - 抽象基类 - 多重继承 - 魔术方法 ### 并发编程 - async/await语法 - 协程和事件循环 - 线程和进程 - 并发数据结构 - 异步上下文管理 ### 元编程 - 动态类创建 - 属性访问控制 - 方法动态绑定 - 代码生成技术 - 反射机制 ## 常见问题 ### 性能问题 - **问题**: 过度使用装饰器导致性能下降 - **原因**: 装饰器增加了函数调用开销 - **解决**: 合理使用装饰器，避免嵌套过深 - **问题**: 生成器使用不当 - **原因**: 不理解生成器的惰性求值特性 - **解决**: 正确理解生成器的工作原理 ### 可读性问题 - **问题**: 过度使用元编程 - **原因**: 代码逻辑过于隐晦 - **解决**: 保持代码简洁，避免过度抽象 - **问题**: 链式调用过长 - **原因**: 代码可读性差 - **解决**: 合理拆分链式调用 ### 兼容性问题 - **问题**: 异步代码兼容性 - **原因**: 不同Python版本的异步特性差异 - **解决**: 检查目标Python版本，使用合适的异步模式 ## 代码示例 ### 装饰器高级应用 ```python # 基础装饰器 def timing_decorator(func): """计时装饰器""" import time import functools @functools.wraps(func) def wrapper(*args, **kwargs): start_time = time.time() result = func(*args, **kwargs) end_time = time.time() print(f"{func.__name__} 执行时间: {end_time - start_time:.4f}秒") return result return wrapper # 带参数的装饰器 def retry(max_attempts=3, delay=1, exceptions=(Exception,)): """重试装饰器""" import time import functools def decorator(func): @functools.wraps(func) def wrapper(*args, **kwargs): last_exception = None for attempt in range(max_attempts): try: return func(*args, **kwargs) except exceptions as e: last_exception = e if attempt < max_attempts - 1: print(f"第 {attempt + 1} 次尝试失败，{delay}秒后重试...") time.sleep(delay) else: print(f"所有 {max_attempts} 次尝试都失败了") raise last_exception return wrapper return decorator # 缓存装饰器 def cache(max_size=128): """LRU缓存装饰器""" import functools from collections import OrderedDict def decorator(func): cache_dict = OrderedDict() @functools.wraps(func) def wrapper(*args, **kwargs): # 创建缓存键 key = str(args) + str(sorted(kwargs.items())) if key in cache_dict: # 移动到末尾（最近使用） cache_dict.move_to_end(key) return cache_dict[key] # 计算结果 result = func(*args, **kwargs) # 添加到缓存 cache_dict[key] = result # 限制缓存大小 if len(cache_dict) > max_size: cache_dict.popitem(last=False) # 移除最久未使用的项 return result # 添加缓存管理方法 def cache_clear(): cache_dict.clear() def cache_info(): return { 'size': len(cache_dict), 'max_size': max_size, 'keys': list(cache_dict.keys()) } wrapper.cache_clear = cache_clear wrapper.cache_info = cache_info return wrapper return decorator # 权限检查装饰器 def permission_required(permission): """权限检查装饰器""" import functools def decorator(func): @functools.wraps(func) def wrapper(self, *args, **kwargs): if not hasattr(self, 'user_permissions'): raise PermissionError("用户权限信息未设置") if permission not in self.user_permissions: raise PermissionError(f"需要权限: {permission}") return func(self, *args, **kwargs) return wrapper return decorator # 使用示例 class UserService: def __init__(self, user_permissions): self.user_permissions = user_permissions @timing_decorator @retry(max_attempts=3, delay=1) @cache(max_size=100) def get_user_data(self, user_id): """获取用户数据""" import random import time # 模拟数据库查询 time.sleep(0.1) # 模拟随机失败 if random.random() < 0.3: raise ConnectionError("数据库连接失败") return { 'id': user_id, 'name': f'用户{user_id}', 'email': f'user{user_id}@example.com' } @permission_required('admin') def delete_user(self, user_id): """删除用户""" return f"用户 {user_id} 已删除" # 装饰器工厂 def validate_types(**type_hints): """类型验证装饰器""" import functools inspect = __import__('inspect').inspect def decorator(func): signature = inspect.signature(func) @functools.wraps(func) def wrapper(*args, **kwargs): # 绑定参数 bound_args = signature.bind(*args, **kwargs) bound_args.apply_defaults() # 验证类型 for param_name, param_value in bound_args.arguments.items(): if param_name in type_hints: expected_type = type_hints[param_name] if not isinstance(param_value, expected_type): raise TypeError( f"参数 {param_name} 应该是 {expected_type.__name__} 类型，" f"但得到的是 {type(param_value).__name__}" ) return func(*args, **kwargs) return wrapper return decorator @validate_types(name=str, age=int, email=str) def create_user(name, age, email): return f"创建用户: {name}, {age}岁, {email}" ``` ### 生成器和迭代器高级应用 ```python # 自定义迭代器 class FibonacciIterator: """斐波那契数列迭代器""" def __init__(self, max_count=None): self.max_count = max_count self.current = 0 self.a, self.b = 0, 1 self.count = 0 def __iter__(self): return self def __next__(self): if self.max_count and self.count >= self.max_count: raise StopIteration if self.count == 0: self.count += 1 return 0 result = self.b self.a, self.b = self.b, self.a + self.b self.count += 1 return result # 生成器函数 def prime_generator(): """素数生成器""" def is_prime(n): if n < 2: return False for i in range(2, int(n ** 0.5) + 1): if n % i == 0: return False return True num = 2 while True: if is_prime(num): yield num num += 1 # 管道生成器 def pipeline_generator(): """数据处理管道生成器""" def read_numbers(filename): """读取数字""" with open(filename, 'r') as f: for line in f: yield int(line.strip()) def filter_even(numbers): """过滤偶数""" for num in numbers: if num % 2 == 0: yield num def multiply_by_two(numbers): """乘以2""" for num in numbers: yield num * 2 def sum_numbers(numbers): """求和""" total = 0 for num in numbers: total += num return total # 组合管道 numbers = read_numbers('numbers.txt') even_numbers = filter_even(numbers) doubled_numbers = multiply_by_two(even_numbers) result = sum_numbers(doubled_numbers) return result # 协程生成器 def coroutine_example(): """协程示例""" def accumulator(): """累加器协程""" total = 0 while True: value = yield total if value is None: break total += value return total def average_calculator(): """平均值计算协程""" count = 0 total = 0 while True: value = yield if value is None: break count += 1 total += value return total / count if count > 0 else 0 # 使用协程 acc = accumulator() next(acc) # 启动协程 print(acc.send(10)) # 10 print(acc.send(20)) # 30 print(acc.send(30)) # 60 try: acc.send(None) # 结束协程 except StopIteration as e: print("累加器结果:", e.value) # 平均值计算 avg = average_calculator() next(avg) for num in [10, 20, 30, 40, 50]: avg.send(num) try: avg.send(None) except StopIteration as e: print("平均值:", e.value) # 上下文管理器生成器 from contextlib import contextmanager @contextmanager def file_manager(filename, mode): """文件管理上下文""" try: f = open(filename, mode) yield f finally: f.close() @contextmanager def database_connection(connection_string): """数据库连接上下文""" import sqlite3 conn = None try: conn = sqlite3.connect(connection_string) yield conn except Exception as e: print(f"数据库错误: {e}") raise finally: if conn: conn.close() # 使用示例 def generator_examples(): # 使用斐波那契迭代器 fib_iter = FibonacciIterator(10) print("前10个斐波那契数:", list(fib_iter)) # 使用素数生成器 prime_gen = prime_generator() first_10_primes = [next(prime_gen) for _ in range(10)] print("前10个素数:", first_10_primes) # 使用上下文管理器 with file_manager('test.txt', 'w') as f: f.write('Hello, World!') with file_manager('test.txt', 'r') as f: content = f.read() print("文件内容:", content) # 协程示例 coroutine_example() ``` ### 元类编程 ```python # 基础元类 class SingletonMeta(type): """单例元类""" _instances = {} def __call__(cls, *args, **kwargs): if cls not in cls._instances: cls._instances[cls] = super().__call__(*args, **kwargs) return cls._instances[cls] class Singleton(metaclass=SingletonMeta): """单例基类""" def __init__(self): self.value = 0 # 属性验证元类 class ValidatedMeta(type): """属性验证元类""" def __new__(cls, name, bases, namespace): # 创建类 new_class = super().__new__(cls, name, bases, namespace) # 添加属性验证 if hasattr(new_class, '_validators'): for attr_name, validator in new_class._validators.items(): setattr(new_class, attr_name, ValidatedMeta.create_validated_property(attr_name, validator)) return new_class @staticmethod def create_validated_property(attr_name, validator): """创建验证属性""" private_name = f'_{attr_name}' def getter(self): return getattr(self, private_name) def setter(self, value): if not validator(value): raise ValueError(f"属性 {attr_name} 验证失败") setattr(self, private_name, value) return property(getter, setter) # ORM元类 class ORMMeta(type): """ORM元类""" def __new__(cls, name, bases, namespace): # 创建类 new_class = super().__new__(cls, name, bases, namespace) # 如果有字段定义，创建表结构 if hasattr(new_class, '_fields'): new_class._table_name = name.lower() new_class._field_definitions = {} for field_name, field_type in new_class._fields.items():