python-patterns

name	python-patterns
description	构建健壮、高效且易于维护的 Python 应用程序的 Python 惯用法（Pythonic idioms）、PEP 8 标准、类型提示（Type hints）以及最佳实践。
origin	ECC

Python 开发模式 (Python Development Patterns)

用于构建健壮、高效且易于维护的应用程序的 Python 惯用模式与最佳实践。

何时激活

• 编写新的 Python 代码时
• 评审 Python 代码时
• 重构现有的 Python 代码时
• 设计 Python 包（Packages）或模块（Modules）时

核心原则

1. 可读性至上 (Readability Counts)

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. 显式优于隐式 (Explicit is Better Than Implicit)

避免使用“魔法”；确保代码的行为清晰透明。

# 推荐：显式配置
import logging

logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)

# 不推荐：隐藏的副作用
import some_module
some_module.setup()  # 这行代码具体做了什么？

3. EAFP - 宽恕好过许可 (Easier to Ask Forgiveness Than Permission)

Python 倾向于使用异常处理而非预先检查条件。

# 推荐：EAFP 风格
def get_value(dictionary: dict, key: str) -> Any:
    try:
        return dictionary[key]
    except KeyError:
        return default_value

# 不推荐：LBYL (Look Before You Leap) 风格
def get_value(dictionary: dict, key: str) -> Any:
    if key in dictionary:
        return dictionary[key]
    else:
        return default_value

类型提示 (Type Hints)

基础类型标注

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"未找到配置文件: {path}") from e
    except json.JSONDecodeError as e:
        raise ConfigError(f"配置文件中的 JSON 无效: {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  # 静默失败！

异常链 (Exception Chaining)

def process_data(data: str) -> Result:
    try:
        parsed = json.loads(data)
    except json.JSONDecodeError as e:
        # 链接异常以保留回溯信息
        raise ValueError(f"无法解析数据: {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_id}")
    return user

上下文管理器 (Context Managers)

资源管理

# 推荐：使用上下文管理器
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} 耗时 {elapsed:.4f} 秒")

# 用法
with timer("数据处理"):
    process_large_dataset()

上下文管理器类

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  # 不要抑制异常

# 用法
with DatabaseTransaction(conn):
    user = conn.create_user(user_data)
    conn.create_profile(user.id, profile_data)

推导式 (Comprehensions) 与生成器 (Generators)

列表推导式 (List Comprehensions)

# 推荐：使用列表推导式进行简单的转换
names = [user.name for user in users if user.is_active]

# 不推荐：手动循环
names = []
for user in users:
    if user.is_active:
        names.append(user.name)

# 复杂的推导式应当展开
# 不推荐：过于复杂
result = [x * 2 for x in items if x > 0 if x % 2 == 0]

# 推荐：使用生成器函数
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

生成器表达式 (Generator Expressions)

# 推荐：使用生成器进行惰性求值
total = sum(x * x for x in range(1_000_000))

# 不推荐：创建大型中间列表
total = sum([x * x for x in range(1_000_000)])

生成器函数

def read_large_file(path: str) -> Iterator[str]:
    """逐行读取大文件。"""
    with open(path) as f:
        for line in f:
            yield line.strip()

# 用法
for line in read_large_file("huge.txt"):
    process(line)

数据类 (Data Classes) 与具名元组 (Named Tuples)

数据类 (Data Classes)

from dataclasses import dataclass, field
from datetime import datetime

@dataclass
class User:
    """具有自动生成的 __init__、__repr__ 和 __eq__ 的用户实体。"""
    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"无效的邮箱: {self.email}")
        # 验证年龄范围
        if self.age < 0 or self.age > 150:
            raise ValueError(f"无效的年龄: {self.age}")

具名元组 (Named Tuples)

from typing import NamedTuple

class Point(NamedTuple):
    """不可变的 2D 点。"""
    x: float
    y: float

    def distance(self, other: 'Point') -> float:
        return ((self.x - other.x) ** 2 + (self.y - other.y) ** 2) ** 0.5

# 用法
p1 = Point(0, 0)
p2 = Point(3, 4)
print(p1.distance(p2))  # 5.0

装饰器 (Decorators)

函数装饰器

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__} 耗时 {elapsed:.4f}s")
        return result
    return wrapper

@timer
def slow_function():
    time.sleep(1)

# slow_function() 打印: slow_function took 1.0012s

参数化装饰器

def repeat(times: int):
    """多次重复执行函数的装饰器。"""
    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") 返回 ["Hello, Alice!", "Hello, Alice!", "Hello, Alice!"]

基于类的装饰器

class CountCalls:
    """计算函数调用次数的装饰器。"""
    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__} 已被调用 {self.count} 次")
        return self.func(*args, **kwargs)

@CountCalls
def process():
    pass

# 每次调用 process() 都会打印调用计数

并发模式 (Concurrency Patterns)

用于 I/O 密集型任务的多线程 (Threading)

import concurrent.futures
import threading

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 密集型任务的多进程 (Multiprocessing)

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

用于并发 I/O 的 Async/Await

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

包结构组织 (Package Organization)

标准项目布局

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

导入规范

# 推荐：导入顺序 - 标准库、第三方库、本地库
import os
import sys
from pathlib import Path

import requests
from fastapi import FastAPI

from mypackage.models import User
from mypackage.utils import format_name

# 推荐：使用 isort 进行自动导入排序
# pip install isort

用于包导出的 init.py

# mypackage/__init__.py
"""mypackage - 一个 Python 包示例。"""

__version__ = "1.0.0"

# 在包级别导出主要的类/函数
from mypackage.models import User, Post
from mypackage.utils import format_name

__all__ = ["User", "Post", "format_name"]

内存与性能

使用 slots 提高内存效率

# 不推荐：普通类使用 __dict__（占用更多内存）
class Point:
    def __init__(self, x: float, y: float):
        self.x = x
        self.y = y

# 推荐：__slots__ 减少内存使用
class Point:
    __slots__ = ['x', 'y']

    def __init__(self, x: float, y: float):
        self.x = x
        self.y = y

用于大数据的生成器

# 不推荐：在内存中返回完整列表
def read_lines(path: str) -> list[str]:
    with open(path) as f:
        return [line.strip() for line in f]

# 推荐：一次产出一行
def read_lines(path: str) -> Iterator[str]:
    with open(path) as f:
        for line in f:
            yield line.strip()

避免在循环中进行字符串拼接

# 不推荐：由于字符串不可变性，导致 O(n²) 复杂度
result = ""
for item in items:
    result += str(item)

# 推荐：使用 join 实现 O(n) 复杂度
result = "".join(str(item) for item in items)

# 推荐：使用 StringIO 进行构建
from io import StringIO

buffer = StringIO()
for item in items:
    buffer.write(str(item))
result = buffer.getvalue()

Python 工具链集成

常用命令

# 代码格式化
black .
isort .

# 静态检查 (Linting)
ruff check .
pylint mypackage/

# 类型检查
mypy .

# 测试
pytest --cov=mypackage --cov-report=html

# 安全扫描
bandit -r .

# 依赖管理
pip-audit
safety check

pyproject.toml 配置

[project]
name = "mypackage"
version = "1.0.0"
requires-python = ">=3.9"
dependencies = [
    "requests>=2.31.0",
    "pydantic>=2.0.0",
]

[project.optional-dependencies]
dev = [
    "pytest>=7.4.0",
    "pytest-cov>=4.1.0",
    "black>=23.0.0",
    "ruff>=0.1.0",
    "mypy>=1.5.0",
]

[tool.black]
line-length = 88
target-version = ['py39']

[tool.ruff]
line-length = 88
select = ["E", "F", "I", "N", "W"]

[tool.mypy]
python_version = "3.9"
warn_return_any = true
warn_unused_configs = true
disallow_untyped_defs = true

[tool.pytest.ini_options]
testpaths = ["tests"]
addopts = "--cov=mypackage --cov-report=term-missing"

快速参考：Python 惯用法

惯用法	描述
EAFP	宽恕好过许可 (Easier to Ask Forgiveness than Permission)
上下文管理器 (Context managers)	使用 with 进行资源管理
列表推导式 (List comprehensions)	用于简单的转换
生成器 (Generators)	用于惰性求值和大型数据集
类型提示 (Type hints)	标注函数签名
数据类 (Dataclasses)	用于带有自动生成方法的资源容器
__slots__	用于内存优化
f-strings	用于字符串格式化 (Python 3.6+)
pathlib.Path	用于路径操作 (Python 3.4+)
enumerate	在循环中获取索引-元素对

应避免的反模式 (Anti-Patterns)

# 不推荐：可变默认参数
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()

# 不推荐：from module import *
from os.path import *

# 推荐：显式导入
from os.path import join, exists

# 不推荐：空 except
try:
    risky_operation()
except:
    pass

# 推荐：特定的异常
try:
    risky_operation()
except SpecificError as e:
    logger.error(f"操作失败: {e}")

记住：Python 代码应当是可读的、显式的，并遵循“最小惊讶原则”（principle of least surprise）。如有疑虑，请优先考虑清晰度而非技巧性。