python-patterns

# Python 开发模式 构建健壮、高效、可维护应用的 Pythonic 模式和最佳实践。 ## 何时激活此技能 - 编写新的 Python 代码 - 审查 Python 代码 - 重构现有 Python 代码 - 设计 Python 包/模块 ## 核心原则 ### 1. 可读性很重要 Python 优先考虑可读性。代码应该明显且易于理解。 ```python # 好的写法：清晰可读 def get_active_users(users: list[User]) -> list[User]: """从提供的列表中返回活跃用户。""" return [user for user in users if user.is_active] # 不好的写法：巧妙但令人困惑 def get_active_users(u): return [x for x in u if x.a] ``` ### 2. 显式优于隐式 避免魔法，清楚说明你的代码做什么。 ```python # 好的写法：显式配置 import logging logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' ) # 不好的写法：隐藏的副作用 import some_module some_module.setup() # 这做什么？ ``` ### 3. EAFP（请求原谅比请求许可更容易） Python 偏好异常处理而非检查条件。 ```python # 好的写法：EAFP 风格 def get_value(dictionary: dict, key: str) -> Any: try: return dictionary[key] except KeyError: return default_value # 不好的写法：LBYL（三思而后行）风格 def get_value(dictionary: dict, key: str) -> Any: if key in dictionary: return dictionary[key] else: return default_value ``` ## 类型提示 ### 基本类型注解 ```python from typing import Optional, List, Dict, Any def process_user( user_id: str, data: Dict[str, Any], active: bool = True ) -> Optional[User]: """处理用户并返回更新的 User 或 None。""" if not active: return None return User(user_id, data) ``` ### 现代类型提示 (Python 3.9+) ```python # Python 3.9+ - 使用内置类型 def process_items(items: list[str]) -> dict[str, int]: return {item: len(item) for item in items} # Python 3.8 及更早版本 - 使用 typing 模块 from typing import List, Dict def process_items(items: List[str]) -> Dict[str, int]: return {item: len(item) for item in items} ``` ### 类型别名和 TypeVar ```python from typing import TypeVar, Union # 复杂类型的类型别名 JSON = Union[dict[str, Any], list[Any], str, int, float, bool, None] def parse_json(data: str) -> JSON: return json.loads(data) # 泛型类型 T = TypeVar('T') def first(items: list[T]) -> T | None: """返回第一个项目，如果列表为空则返回 None。""" return items[0] if items else None ``` ### 基于 Protocol 的鸭子类型 ```python from typing import Protocol class Renderable(Protocol): def render(self) -> str: """将对象渲染为字符串。""" def render_all(items: list[Renderable]) -> str: """渲染所有实现 Renderable 协议的项目。""" return "\n".join(item.render() for item in items) ``` ## 错误处理模式 ### 特定异常处理 ```python # 好的写法：捕获特定异常 def load_config(path: str) -> Config: try: with open(path) as f: return Config.from_json(f.read()) except FileNotFoundError as e: raise ConfigError(f"Config file not found: {path}") from e except json.JSONDecodeError as e: raise ConfigError(f"Invalid JSON in config: {path}") from e # 不好的写法：裸 except def load_config(path: str) -> Config: try: with open(path) as f: return Config.from_json(f.read()) except: return None # 静默失败！ ``` ### 异常链 ```python def process_data(data: str) -> Result: try: parsed = json.loads(data) except json.JSONDecodeError as e: # 链接异常以保留回溯 raise ValueError(f"Failed to parse data: {data}") from e ``` ### 自定义异常层次结构 ```python class AppError(Exception): """所有应用错误的基异常。""" pass class ValidationError(AppError): """输入验证失败时抛出。""" pass class NotFoundError(AppError): """请求的资源未找到时抛出。""" pass # 使用方式 def get_user(user_id: str) -> User: user = db.find_user(user_id) if not user: raise NotFoundError(f"User not found: {user_id}") return user ``` ## 上下文管理器 ### 资源管理 ```python # 好的写法：使用上下文管理器 def process_file(path: str) -> str: with open(path, 'r') as f: return f.read() # 不好的写法：手动资源管理 def process_file(path: str) -> str: f = open(path, 'r') try: return f.read() finally: f.close() ``` ### 自定义上下文管理器 ```python from contextlib import contextmanager @contextmanager def timer(name: str): """计时代码块的上下文管理器。""" start = time.perf_counter() yield elapsed = time.perf_counter() - start print(f"{name} took {elapsed:.4f} seconds") # 使用方式 with timer("data processing"): process_large_dataset() ``` ## 数据类和命名元组 ### 数据类 自动生成 `__init__`、`__repr__` 和 `__eq__`。 ```python from dataclasses import dataclass, field from datetime import datetime @dataclass class User: """带有自动生成方法的数据容器。""" id: str name: str email: str created_at: datetime = field(default_factory=datetime.now) is_active: bool = True # 使用方式 user = User( id="123", name="Alice", email="alice@example.com" ) ``` ### 带验证的数据类 ```python @dataclass class User: email: str age: int def __post_init__(self): # 验证邮箱格式 if "@" not in self.email: raise ValueError(f"Invalid email: {self.email}") # 验证年龄范围 if self.age < 0 or self.age > 150: raise ValueError(f"Invalid age: {self.age}") ``` ## 装饰器 ### 函数装饰器 ```python import functools import time def timer(func: Callable) -> Callable: """计时函数执行的装饰器。""" @functools.wraps(func) def wrapper(*args, **kwargs): start = time.perf_counter() result = func(*args, **kwargs) elapsed = time.perf_counter() - start print(f"{func.__name__} took {elapsed:.4f}s") return result return wrapper @timer def slow_function(): time.sleep(1) ``` ## 并发模式 ### I/O 密集型任务使用线程 ```python import concurrent.futures def fetch_url(url: str) -> str: """获取 URL（I/O 密集型操作）。""" import urllib.request with urllib.request.urlopen(url) as response: return response.read().decode() def fetch_all_urls(urls: list[str]) -> dict[str, str]: """使用线程并发获取多个 URL。""" with concurrent.futures.ThreadPoolExecutor(max_workers=10) as executor: future_to_url = {executor.submit(fetch_url, url): url for url in urls} results = {} for future in concurrent.futures.as_completed(future_to_url): url = future_to_url[future] try: results[url] = future.result() except Exception as e: results[url] = f"Error: {e}" return results ``` ### CPU 密集型任务使用多进程 ```python def process_data(data: list[int]) -> int: """CPU 密集型计算。""" return sum(x ** 2 for x in data) def process_all(datasets: list[list[int]]) -> list[int]: """使用多进程处理多个数据集。""" with concurrent.futures.ProcessPoolExecutor() as executor: results = list(executor.map(process_data, datasets)) return results ``` ### Async/Await 用于并发 I/O ```python import asyncio async def fetch_async(url: str) -> str: """异步获取 URL。""" import aiohttp async with aiohttp.ClientSession() as session: async with session.get(url) as response: return await response.text() async def fetch_all(urls: list[str]) -> dict[str, str]: """并发获取多个 URL。""" tasks = [fetch_async(url) for url in urls] results = await asyncio.gather(*tasks, return_exceptions=True) return dict(zip(urls, results)) ``` ## 包组织 ### 标准项目布局 ``` myproject/ ├── src/ │ └── mypackage/ │ ├── __init__.py │ ├── main.py │ ├── api/ │ │ ├── __init__.py │ │ └── routes.py │ ├── models/ │ │ ├── __init__.py │ │ └── user.py │ └── utils/ │ ├── __init__.py │ └── helpers.py ├── tests/ │ ├── __init__.py │ ├── conftest.py │ ├── test_api.py │ └── test_models.py ├── pyproject.toml ├── README.md └── .gitignore ``` ## 避免的反模式 ```python # 不好的写法：可变默认参数 def append_to(item, items=[]): items.append(item) return items # 好的写法：使用 None 并创建新列表 def append_to(item, items=None): if items is None: items = [] items.append(item) return items # 不好的写法：用 type() 检查类型 if type(obj) == list: process(obj) # 好的写法：使用 isinstance if isinstance(obj, list): process(obj) # 不好的写法：用 == 比较 None if value == None: process() # 好的写法：使用 is if value is None: process() # 不好的写法：裸 except try: risky_operation() except: pass # 好的写法：特定异常 try: risky_operation() except SpecificError as e: logger.error(f"Operation failed: {e}") ``` **记住**：Python 代码应该是可读的、显式的，并遵循最小惊讶原则。有疑问时，优先考虑清晰而非巧妙。