python-patterns

# Python 开发模式 用于构建健壮、高效和可维护应用程序的惯用 Python 模式与最佳实践。 ## 何时激活 * 编写新的 Python 代码 * 审查 Python 代码 * 重构现有的 Python 代码 * 设计 Python 包/模块 ## 核心原则 ### 1. 可读性很重要 Python 优先考虑可读性。代码应该清晰且易于理解。 ```python # Good: Clear and readable def get_active_users(users: list[User]) -> list[User]: """Return only active users from the provided list.""" return [user for user in users if user.is_active] # Bad: Clever but confusing def get_active_users(u): return [x for x in u if x.a] ``` ### 2. 显式优于隐式 避免魔法；清晰说明你的代码在做什么。 ```python # Good: Explicit configuration import logging logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' ) # Bad: Hidden side effects import some_module some_module.setup() # What does this do? ``` ### 3. EAFP - 请求宽恕比请求许可更容易 Python 倾向于使用异常处理而非检查条件。 ```python # Good: EAFP style def get_value(dictionary: dict, key: str) -> Any: try: return dictionary[key] except KeyError: return default_value # Bad: LBYL (Look Before You Leap) style 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]: """Process a user and return the updated User or None.""" if not active: return None return User(user_id, data) ``` ### 现代类型提示（Python 3.9+） ```python # Python 3.9+ - Use built-in types def process_items(items: list[str]) -> dict[str, int]: return {item: len(item) for item in items} # Python 3.8 and earlier - Use typing module 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 # Type alias for complex types JSON = Union[dict[str, Any], list[Any], str, int, float, bool, None] def parse_json(data: str) -> JSON: return json.loads(data) # Generic types T = TypeVar('T') def first(items: list[T]) -> T | None: """Return the first item or None if list is empty.""" return items[0] if items else None ``` ### 基于协议的鸭子类型 ```python from typing import Protocol class Renderable(Protocol): def render(self) -> str: """Render the object to a string.""" def render_all(items: list[Renderable]) -> str: """Render all items that implement the Renderable protocol.""" return "\n".join(item.render() for item in items) ``` ## 错误处理模式 ### 特定异常处理 ```python # Good: Catch specific exceptions 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 # Bad: Bare except def load_config(path: str) -> Config: try: with open(path) as f: return Config.from_json(f.read()) except: return None # Silent failure! ``` ### 异常链 ```python def process_data(data: str) -> Result: try: parsed = json.loads(data) except json.JSONDecodeError as e: # Chain exceptions to preserve the traceback raise ValueError(f"Failed to parse data: {data}") from e ``` ### 自定义异常层次结构 ```python class AppError(Exception): """Base exception for all application errors.""" pass class ValidationError(AppError): """Raised when input validation fails.""" pass class NotFoundError(AppError): """Raised when a requested resource is not found.""" pass # Usage 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 # Good: Using context managers def process_file(path: str) -> str: with open(path, 'r') as f: return f.read() # Bad: Manual resource management 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): """Context manager to time a block of code.""" start = time.perf_counter() yield elapsed = time.perf_counter() - start print(f"{name} took {elapsed:.4f} seconds") # Usage with timer("data processing"): process_large_dataset() ``` ### 上下文管理器类 ```python class DatabaseTransaction: def __init__(self, connection): self.connection = connection def __enter__(self): self.connection.begin_transaction() return self def __exit__(self, exc_type, exc_val, exc_tb): if exc_type is None: self.connection.commit() else: self.connection.rollback() return False # Don't suppress exceptions # Usage with DatabaseTransaction(conn): user = conn.create_user(user_data) conn.create_profile(user.id, profile_data) ``` ## 推导式和生成器 ### 列表推导式 ```python # Good: List comprehension for simple transformations names = [user.name for user in users if user.is_active] # Bad: Manual loop names = [] for user in users: if user.is_active: names.append(user.name) # Complex comprehensions should be expanded # Bad: Too complex result = [x * 2 for x in items if x > 0 if x % 2 == 0] # Good: Use a generator function def filter_and_transform(items: Iterable[int]) -> list[int]: result = [] for x in items: if x > 0 and x % 2 == 0: result.append(x * 2) return result ``` ### 生成器表达式 ```python # Good: Generator for lazy evaluation total = sum(x * x for x in range(1_000_000)) # Bad: Creates large intermediate list total = sum([x * x for x in range(1_000_000)]) ``` ### 生成器函数 ```python def read_large_file(path: str) -> Iterator[str]: """Read a large file line by line.""" with open(path) as f: for line in f: yield line.strip() # Usage for line in read_large_file("huge.txt"): process(line) ``` ## 数据类和命名元组 ### 数据类 ```python from dataclasses import dataclass, field from datetime import datetime @dataclass class User: """User entity with automatic __init__, __repr__, and __eq__.""" id: str name: str email: str created_at: datetime = field(default_factory=datetime.now) is_active: bool = True # Usage user = User( id="123", name="Alice", email="alice@example.com" ) ``` ### 带验证的数据类 ```python @dataclass class User: email: str age: int def __post_init__(self): # Validate email format if "@" not in self.email: raise ValueError(f"Invalid email: {self.email}") # Validate age range if self.age < 0 or self.age > 150: raise ValueError(f"Invalid age: {self.age}") ``` ### 命名元组 ```python from typing import NamedTuple class Point(NamedTuple): """Immutable 2D point.""" x: float y: float def distance(self, other: 'Point') -> float: return ((self.x - other.x) ** 2 + (self.y - other.y) ** 2) ** 0.5 # Usage p1 = Point(0, 0) p2 = Point(3, 4) print(p1.distance(p2)) # 5.0 ``` ## 装饰器 ### 函数装饰器 ```python import functools import time def timer(func: Callable) -> Callable: """Decorator to time function execution.""" @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) # slow_function() prints: slow_function took 1.0012s ``` ### 参数化装饰器 ```python def repeat(times: int): """Decorator to repeat a function multiple times.""" def decorator(func: Callable) -> Callable: @functools.wraps(func) def wrapper(*args, **kwargs): results = [] for _ in range(times): results.append(func(*args, **kwargs)) return results return wrapper return decorator @repeat(times=3) def greet(name: str) -> str: return f"Hello, {name}!" # greet("Alice") returns ["Hello, Alice!", "Hello, Alice!", "Hello, Alice!"] ``` ### 基于类的装饰器 ```python class CountCalls: """Decorator that counts how many times a function is called.""" def __init__(self, func: Callable): functools.update_wrapper(self, func) self.func = func self.count = 0 def __call__(self, *args, **kwargs): self.count += 1 print(f"{self.func.__name__} has been called {self.count} times") return self.func(*args, **kwargs) @CountCalls def process(): pass # Each call to process() prints the call count ``` ## 并发模式 ### 用于 I/O 密集型任务的线程 ```python import concurrent.futures import threading def fetch_url(url: str) -> str: """Fetch a URL (I/O-bound operation).""" import urllib.request with urllib.request.urlopen(url) as response: return response.read().decode() def fetch_all_urls(urls: list[str]) -> dict[str, str]: """Fetch multiple URLs concurrently using threads.""" 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-intensive computation.""" return sum(x ** 2 for x in data) def process_all(datasets: list[list[int]]) -> list[int]: """Process multiple datasets using multiple processes.""" with concurrent.futures.ProcessPoolExecutor() as executor: results = list(executor.map(process_data, datasets)) return results ``` ### 用于并发 I/O 的异步/等待 ```python import asyncio async def fetch_async(url: str) -> str: """Fetch a URL asynchronously.""" 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]: """Fetch multiple URLs concurrently.""" tasks = [fetch_async(url) for url in urls]