python-patterns

name	python-patterns
description	Pythonic 惯用语、PEP 8 标准、类型提示和构建健壮、高效、可维护的 Python 应用的最佳实践。

Python 开发模式

构建健壮、高效、可维护应用的 Pythonic 模式和最佳实践。

何时激活此技能

• 编写新的 Python 代码
• 审查 Python 代码
• 重构现有 Python 代码
• 设计 Python 包/模块

核心原则

1. 可读性很重要

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. 显式优于隐式

避免魔法，清楚说明你的代码做什么。

# 好的写法：显式配置
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 偏好异常处理而非检查条件。

# 好的写法：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

类型提示

基本类型注解

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 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

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 的鸭子类型

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)

错误处理模式

特定异常处理

# 好的写法：捕获特定异常
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  # 静默失败！

异常链

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

自定义异常层次结构

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

上下文管理器

资源管理

# 好的写法：使用上下文管理器
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()

自定义上下文管理器

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__。

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="[e-mail]"
)

带验证的数据类

@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}")

装饰器

函数装饰器

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 密集型任务使用线程

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 密集型任务使用多进程

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

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

避免的反模式

# 不好的写法：可变默认参数
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 代码应该是可读的、显式的，并遵循最小惊讶原则。有疑问时，优先考虑清晰而非巧妙。