🚀 快速安装

复制以下命令并运行,立即安装此 Skill:

npx skills add https://github.com/wshobson/agents --skill python-design-patterns

💡 提示:需要 Node.js 和 NPM

Python 设计模式

使用基础设计原则编写可维护的 Python 代码。这些模式帮助您构建易于理解、测试和修改的系统。

何时使用此技能

  • 设计新组件或服务
  • 重构复杂或混乱的代码
  • 决定是否创建抽象
  • 在继承和组合之间进行选择
  • 评估代码复杂性和耦合度
  • 规划模块化架构

核心概念

1. KISS(保持简单)

选择最简单的可行方案。复杂性必须有具体需求作为理由。

2. 单一职责原则

每个单元应该只有一个改变的理由。将关注点分离到专注的组件中。

3. 组合优于继承

通过组合对象来构建行为,而不是扩展类。

4. 三次原则

等到有三个实例后再进行抽象。重复通常比过早抽象更好。

快速开始

# 简单优于巧妙
# 不要使用工厂/注册表模式:
FORMATTERS = {"json": JsonFormatter, "csv": CsvFormatter}

def get_formatter(name: str) -> Formatter:
    return FORMATTERS[name]()

基础模式

模式 1:KISS – 保持简单

在添加复杂性之前,请问:更简单的解决方案是否可行?

# 过度设计:带注册功能的工厂
class OutputFormatterFactory:
    _formatters: dict[str, type[Formatter]] = {}

    @classmethod
    def register(cls, name: str):
        def decorator(formatter_cls):
            cls._formatters[name] = formatter_cls
            return formatter_cls
        return decorator

    @classmethod
    def create(cls, name: str) -> Formatter:
        return cls._formatters[name]()

@OutputFormatterFactory.register("json")
class JsonFormatter(Formatter):
    ...

# 简单:只使用字典
FORMATTERS = {
    "json": JsonFormatter,
    "csv": CsvFormatter,
    "xml": XmlFormatter,
}

def get_formatter(name: str) -> Formatter:
    """根据名称获取格式化器。"""
    if name not in FORMATTERS:
        raise ValueError(f"未知格式:{name}")
    return FORMATTERS[name]()

工厂模式在这里增加了代码但没有增加价值。只有在解决实际问题时才使用模式。

模式 2:单一职责原则

每个类或函数应该只有一个改变的理由。

# 不好:处理器做所有事情
class UserHandler:
    async def create_user(self, request: Request) -> Response:
        # HTTP 解析
        data = await request.json()

        # 验证
        if not data.get("email"):
            return Response({"error": "email required"}, status=400)

        # 数据库访问
        user = await db.execute(
            "INSERT INTO users (email, name) VALUES ($1, $2) RETURNING *",
            data["email"], data["name"]
        )

        # 响应格式化
        return Response({"id": user.id, "email": user.email}, status=201)

# 好:分离关注点
class UserService:
    """仅包含业务逻辑。"""

    def __init__(self, repo: UserRepository) -> None:
        self._repo = repo

    async def create_user(self, data: CreateUserInput) -> User:
        # 这里只包含业务规则
        user = User(email=data.email, name=data.name)
        return await self._repo.save(user)

class UserHandler:
    """仅处理 HTTP 相关事宜。"""

    def __init__(self, service: UserService) -> None:
        self._service = service

    async def create_user(self, request: Request) -> Response:
        data = CreateUserInput(**(await request.json()))
        user = await self._service.create_user(data)
        return Response(user.to_dict(), status=201)

现在,HTTP 变更不会影响业务逻辑,反之亦然。

模式 3:关注点分离

将代码组织成具有明确职责的不同层。

┌─────────────────────────────────────────────────────┐
│  API 层 (处理器)                                      │
│  - 解析请求                                            │
│  - 调用服务                                            │
│  - 格式化响应                                          │
└─────────────────────────────────────────────────────┘
┌─────────────────────────────────────────────────────┐
│ 服务层 (业务逻辑)                                      │
│  - 领域规则和验证                                      │
│  - 编排操作                                            │
│  - 尽可能使用纯函数                                    │
└─────────────────────────────────────────────────────┘
┌─────────────────────────────────────────────────────┐
│ 仓储层 (数据访问)                                      │
│  - SQL 查询                                           │
│  - 外部 API 调用                                      │
│  - 缓存操作                                            │
└─────────────────────────────────────────────────────┘

每一层仅依赖于其下方的层:

# 仓储:数据访问
class UserRepository:
    async def get_by_id(self, user_id: str) -> User | None:
        row = await self._db.fetchrow(
            "SELECT * FROM users WHERE id = $1", user_id
        )
        return User(**row) if row else None

# 服务:业务逻辑
class UserService:
    def __init__(self, repo: UserRepository) -> None:
        self._repo = repo

    async def get_user(self, user_id: str) -> User:
        user = await self._repo.get_by_id(user_id)
        if user is None:
            raise UserNotFoundError(user_id)
        return user

# 处理器:HTTP 相关事宜
@app.get("/users/{user_id}")
async def get_user(user_id: str) -> UserResponse:
    user = await user_service.get_user(user_id)
    return UserResponse.from_user(user)

模式 4:组合优于继承

通过组合对象而不是继承来构建行为。

# 继承:僵化且难以测试
class EmailNotificationService(NotificationService):
    def __init__(self):
        super().__init__()
        self._smtp = SmtpClient()  # 难以模拟

    def notify(self, user: User, message: str) -> None:
        self._smtp.send(user.email, message)

# 组合:灵活且可测试
class NotificationService:
    """通过多个渠道发送通知。"""

    def __init__(
        self,
        email_sender: EmailSender,
        sms_sender: SmsSender | None = None,
        push_sender: PushSender | None = None,
    ) -> None:
        self._email = email_sender
        self._sms = sms_sender
        self._push = push_sender

    async def notify(
        self,
        user: User,
        message: str,
        channels: set[str] | None = None,
    ) -> None:
        channels = channels or {"email"}

        if "email" in channels:
            await self._email.send(user.email, message)

        if "sms" in channels and self._sms and user.phone:
            await self._sms.send(user.phone, message)

        if "push" in channels and self._push and user.device_token:
            await self._push.send(user.device_token, message)

# 使用假对象易于测试
service = NotificationService(
    email_sender=FakeEmailSender(),
    sms_sender=FakeSmsSender(),
)

高级模式

模式 5:三次原则

等到有三个实例后再进行抽象。

# 两个相似函数?先不要抽象
def process_orders(orders: list[Order]) -> list[Result]:
    results = []
    for order in orders:
        validated = validate_order(order)
        result = process_validated_order(validated)
        results.append(result)
    return results

def process_returns(returns: list[Return]) -> list[Result]:
    results = []
    for ret in returns:
        validated = validate_return(ret)
        result = process_validated_return(validated)
        results.append(result)
    return results

# 这些看起来相似,但等等!它们真的相同吗?
# 不同的验证、不同的处理、不同的错误……
# 重复通常比错误的抽象要好

# 只有出现第三个案例后,才考虑是否存在真正的模式
# 但即便如此,有时显式优于抽象

模式 6:函数大小指南

保持函数专注。当函数出现以下情况时进行提取:

  • 超过 20-50 行(取决于复杂性)
  • 服务于多个不同的目的
  • 具有深度嵌套的逻辑(3 级以上)
# 太长,混合了多个关注点
def process_order(order: Order) -> Result:
    # 50 行验证...
    # 30 行库存检查...
    # 40 行支付处理...
    # 20 行通知...
    pass

# 更好:由专注的函数组成
def process_order(order: Order) -> Result:
    """通过完整工作流处理客户订单。"""
    validate_order(order)
    reserve_inventory(order)
    payment_result = charge_payment(order)
    send_confirmation(order, payment_result)
    return Result(success=True, order_id=order.id)

模式 7:依赖注入

通过构造函数传递依赖,以便于测试。

from typing import Protocol

class Logger(Protocol):
    def info(self, msg: str, **kwargs) -> None: ...
    def error(self, msg: str, **kwargs) -> None: ...

class Cache(Protocol):
    async def get(self, key: str) -> str | None: ...
    async def set(self, key: str, value: str, ttl: int) -> None: ...

class UserService:
    """注入依赖的服务。"""

    def __init__(
        self,
        repository: UserRepository,
        cache: Cache,
        logger: Logger,
    ) -> None:
        self._repo = repository
        self._cache = cache
        self._logger = logger

    async def get_user(self, user_id: str) -> User:
        # 先检查缓存
        cached = await self._cache.get(f"user:{user_id}")
        if cached:
            self._logger.info("缓存命中", user_id=user_id)
            return User.from_json(cached)

        # 从数据库获取
        user = await self._repo.get_by_id(user_id)
        if user:
            await self._cache.set(f"user:{user_id}", user.to_json(), ttl=300)

        return user

# 生产环境
service = UserService(
    repository=PostgresUserRepository(db),
    cache=RedisCache(redis),
    logger=StructlogLogger(),
)

# 测试环境
service = UserService(
    repository=InMemoryUserRepository(),
    cache=FakeCache(),
    logger=NullLogger(),
)

模式 8:避免常见反模式

不要暴露内部类型:

# 不好:将 ORM 模型暴露给 API
@app.get("/users/{id}")
def get_user(id: str) -> UserModel:  # SQLAlchemy 模型
    return db.query(UserModel).get(id)

# 好:使用响应模式
@app.get("/users/{id}")
def get_user(id: str) -> UserResponse:
    user = db.query(UserModel).get(id)
    return UserResponse.from_orm(user)

不要将 I/O 与业务逻辑混合:

# 不好:SQL 嵌入在业务逻辑中
def calculate_discount(user_id: str) -> float:
    user = db.query("SELECT * FROM users WHERE id = ?", user_id)
    orders = db.query("SELECT * FROM orders WHERE user_id = ?", user_id)
    # 业务逻辑与数据访问混合

# 好:仓储模式
def calculate_discount(user: User, order_history: list[Order]) -> float:
    # 纯业务逻辑,易于测试
    if len(order_history) > 10:
        return 0.15
    return 0.0

最佳实践总结

  1. 保持简单 – 选择最简单的可行方案
  2. 单一职责 – 每个单元只有一个改变的理由
  3. 关注点分离 – 具有明确目的的不同层
  4. 组合而非继承 – 组合对象以获得灵活性
  5. 三次原则 – 在抽象之前等待
  6. 保持函数小巧 – 20-50 行(取决于复杂性),一个目的
  7. 注入依赖 – 构造函数注入以提高可测试性
  8. 删除先于抽象 – 删除死代码,然后再考虑模式
  9. 测试每一层 – 对每个关注点进行隔离测试
  10. 显式优于巧妙 – 可读的代码胜于优雅的代码

📄 原始文档

完整文档(英文):

https://skills.sh/wshobson/agents/python-design-patterns

💡 提示:点击上方链接查看 skills.sh 原始英文文档,方便对照翻译。

声明:本站所有文章,如无特殊说明或标注,均为本站原创发布。任何个人或组织,在未征得本站同意时,禁止复制、盗用、采集、发布本站内容到任何网站、书籍等各类媒体平台。如若本站内容侵犯了原著者的合法权益,可联系我们进行处理。