chain-of-responsibility-python

Compare original and translation side by side

🇺🇸

Original

English

🇨🇳

Translation

Chinese

Chain of Responsibility (Python)

Chain of Responsibility（Python实现）

Why: Chain of Responsibility lets you pass a request (or context) along a chain of handlers. Each handler decides whether to process it and pass to the next, or short-circuit. You avoid one big function with all steps and keep each step in its own class (Refactoring.Guru).

Hard constraints: Handlers share a single interface (e.g.

handle(context)

). Each handler holds a reference to the next; the client composes the chain. A handler either processes and passes, or passes without processing.

适用原因：Chain of Responsibility模式允许你将请求（或上下文）沿着处理器链传递。每个处理器可以决定是处理请求后传递给下一个，还是直接终止流程。这样你就可以避免编写一个包含所有步骤的庞大函数，而是将每个步骤封装在独立的类中（参考Refactoring.Guru）。

硬性约束：所有处理器共享同一个接口（例如

handle(context)

）。每个处理器持有下一个处理器的引用；由客户端来构建这条链。处理器要么处理后传递，要么直接传递而不处理。

When to use

适用场景

Validation: Multi-rule validation (required → format → range) where you want to add or reorder rules without editing a single validator.
Any sequential pipeline: Processing steps, transformation chains, or multi-step checks where order matters and each step can process and pass (or stop).
You want to decouple the sender from concrete handlers and add or reorder steps without changing existing code (Single Responsibility; Open/Closed).

验证场景：多规则验证（必填检查 → 格式检查 → 范围检查），此时你希望在不修改单个验证器代码的情况下，添加或重新排序验证规则。
任意顺序流水线：处理步骤、转换链或多步骤检查，这些场景中顺序很重要，且每个步骤可以选择处理后传递（或终止）。
你希望将请求发送者与具体处理器解耦，并且可以在不修改现有代码的情况下添加或重新排序步骤（遵循单一职责原则和开闭原则）。

Structure

结构

Role	Responsibility
Handler (protocol/ABC)	Declares `handle(context)` (and optionally `set_next(next)` ). All concrete handlers implement this.
Base handler (optional)	Holds `_next` reference; default `handle()` forwards to `_next` if present. Reduces boilerplate.
Concrete handlers	Implement `handle()` . Process the context (e.g. add errors, transform, check); call `self._next.handle(context)` or return.
Client	Builds the chain (e.g. `a.set_next(b).set_next(c)` ) and invokes the first handler with the initial context.

A context object is passed through the chain; handlers read it, optionally mutate it, and pass it along (e.g. for validation:

value

field_name

errors

角色	职责
Handler（协议/ABC）	声明 `handle(context)` 方法（可选 `set_next(next)` 方法）。所有具体处理器都实现该接口。
基础处理器（可选）	持有 `_next` 引用；默认的 `handle()` 方法会在存在下一个处理器时将请求转发给它。减少重复代码。
具体处理器	实现 `handle()` 方法。处理上下文（例如添加错误信息、转换数据、执行检查）；调用 `self._next.handle(context)` 或直接返回。
客户端	构建处理器链（例如 `a.set_next(b).set_next(c)` ），并使用初始上下文调用第一个处理器。

上下文对象会在链中传递；处理器可以读取它，可选地修改它，然后继续传递（例如在验证场景中，上下文包含

value

、

field_name

、

errors

）。

Code contrast (validation example)

代码对比（验证示例）

Validation is a common use; the same structure applies to any chain. Below: validation.

验证是该模式的常见应用场景；同样的结构也适用于其他类型的链。以下是验证场景的代码对比：

❌ ANTI-PATTERN: One function with all rules

❌ 反模式：单个函数包含所有规则

python

undefined

python

undefined

One function; every new rule forces edits.

单个函数；每次添加新规则都需要修改该函数。

def validate_order_input(data: OrderInput) -> list[dict]: errors = [] if not (data.email or "").strip(): errors.append({"field": "email", "message": "email is required"}) elif not re.match(r"^[^@]+@[^@]+.\w+$", data.email): errors.append({"field": "email", "message": "invalid email"}) if data.amount is None or not (1 <= data.amount <= 10000): errors.append({"field": "amount", "message": "amount must be 1-10000"}) return errors


Problems: order and logic are hardcoded; adding/removing a rule touches this function; rules are hard to test in isolation; violates Open/Closed.


问题：顺序和逻辑硬编码；添加/移除规则必须修改该函数；规则难以单独测试；违反开闭原则。

✅ TOP-CODER PATTERN: Validator protocol + base handler + concrete validators + client-built chain

✅ 推荐模式：验证器协议 + 基础处理器 + 具体验证器 + 客户端构建的链

Validator protocol and context:

python

undefined

验证器协议和上下文：

python

undefined

validators/base.py

from typing import Protocol, runtime_checkable, Optional from dataclasses import field, dataclass

@runtime_checkable class Validator(Protocol): def validate(self, context: "ValidationContext") -> None: ... def set_next(self, next_validator: "Validator") -> "Validator": ...

@dataclass class ValidationContext: value: any field_name: str errors: list = field(default_factory=list)

class BaseValidator: def init(self) -> None: self._next: Optional[Validator] = None

def set_next(self, next_validator: Validator) -> Validator:
    self._next = next_validator
    return next_validator

def validate(self, context: ValidationContext) -> None:
    if self._next:
        self._next.validate(context)


**Concrete validators** (each does one check, then passes):

```python

from typing import Protocol, runtime_checkable, Optional from dataclasses import field, dataclass

@runtime_checkable class Validator(Protocol): def validate(self, context: "ValidationContext") -> None: ... def set_next(self, next_validator: "Validator") -> "Validator": ...

@dataclass class ValidationContext: value: any field_name: str errors: list = field(default_factory=list)

class BaseValidator: def init(self) -> None: self._next: Optional[Validator] = None

def set_next(self, next_validator: Validator) -> Validator:
    self._next = next_validator
    return next_validator

def validate(self, context: ValidationContext) -> None:
    if self._next:
        self._next.validate(context)


**具体验证器**（每个验证器只执行一项检查，然后传递）：

```python

validators/required.py

from .base import BaseValidator, ValidationContext

class RequiredValidator(BaseValidator): def validate(self, context: ValidationContext) -> None: if context.value is None or str(context.value).strip() == "": context.errors.append({ "field": context.field_name, "message": f"{context.field_name} is required", }) super().validate(context)

from .base import BaseValidator, ValidationContext

validators/email_format.py

import re EMAIL_RE = re.compile(r"^[^@]+@[^@]+.\w+$")

class EmailFormatValidator(BaseValidator): def validate(self, context: ValidationContext) -> None: if context.value and not EMAIL_RE.match(str(context.value)): context.errors.append({"field": context.field_name, "message": "invalid email format"}) super().validate(context)

import re EMAIL_RE = re.compile(r"^[^@]+@[^@]+.\w+$")

validators/range.py

class RangeValidator(BaseValidator): def init(self, min_val: float, max_val: float) -> None: super().init() self.min_val = min_val self.max_val = max_val

def validate(self, context: ValidationContext) -> None:
    try:
        n = float(context.value) if context.value is not None else None
    except (TypeError, ValueError):
        n = None
    if context.value is not None and (n is None or n < self.min_val or n > self.max_val):
        context.errors.append({
            "field": context.field_name,
            "message": f"must be between {self.min_val} and {self.max_val}",
        })
    super().validate(context)


**Client** (e.g. service or FastAPI dependency) builds one chain per field and runs it:

```python

class RangeValidator(BaseValidator): def init(self, min_val: float, max_val: float) -> None: super().init() self.min_val = min_val self.max_val = max_val

def validate(self, context: ValidationContext) -> None:
    try:
        n = float(context.value) if context.value is not None else None
    except (TypeError, ValueError):
        n = None
    if context.value is not None and (n is None or n < self.min_val or n > self.max_val):
        context.errors.append({
            "field": context.field_name,
            "message": f"must be between {self.min_val} and {self.max_val}",
        })
    super().validate(context)


**客户端**（例如服务或FastAPI依赖）为每个字段构建一条链并执行：

```python

services/order_validation.py

from validators.required import RequiredValidator from validators.email_format import EmailFormatValidator from validators.range import RangeValidator from validators.base import ValidationContext

email_chain = RequiredValidator() email_chain.set_next(EmailFormatValidator())

amount_chain = RequiredValidator() amount_chain.set_next(RangeValidator(1, 10_000))

def validate_order_input(data: OrderInput) -> list[dict]: ctx = ValidationContext(value=None, field_name="", errors=[]) ctx.value, ctx.field_name = data.email, "email" email_chain.validate(ctx) ctx.value, ctx.field_name = data.amount, "amount" amount_chain.validate(ctx) return ctx.errors


Benefits: add or reorder validators by composing the chain; each validator is a single class, easy to unit test.

---

from validators.required import RequiredValidator from validators.email_format import EmailFormatValidator from validators.range import RangeValidator from validators.base import ValidationContext

email_chain = RequiredValidator() email_chain.set_next(EmailFormatValidator())

amount_chain = RequiredValidator() amount_chain.set_next(RangeValidator(1, 10_000))


优势：通过组合链来添加或重新排序验证器；每个验证器都是独立的类，易于单元测试。

---

Python notes

Python相关注意事项

Validation context: Use a dataclass or simple class with
```
value
```
,
```
field_name
```
, and
```
errors: list
```
so validators append to the same list. For fail-fast, skip calling
```
super().validate(context)
```
when a validator adds an error.
Protocol vs ABC: Prefer
```
typing.Protocol
```
for the validator contract; use
```
BaseValidator
```
only for the common “forward to next” logic.
Chaining:
```
set_next
```
can return the next validator so the client can write
```
required.set_next(email_format).set_next(amount_range)
```
.
Modules: One module per validator when logic is non-trivial (e.g.
```
validators/email_format.py
```
); keep protocol and base in
```
validators/base.py
```
.
No overkill: For one or two fixed steps, a simple function may be enough; use CoR when you have many steps or dynamic composition.
General chains: Same pattern works for non-validation pipelines (e.g. data transformation, enrichment, multi-step processing)—use a context that fits the domain and handlers that process and pass.

验证上下文：使用数据类或简单类存储
```
value
```
、
```
field_name
```
和
```
errors: list
```
，以便所有验证器都能向同一个列表中添加错误信息。如果需要快速失败（fail-fast），当验证器添加错误信息后，跳过调用
```
super().validate(context)
```
即可。
Protocol vs ABC：优先使用
```
typing.Protocol
```
来定义验证器契约；仅在需要共享“转发给下一个处理器”的通用逻辑时使用
```
BaseValidator
```
。
链式调用：
```
set_next
```
方法可以返回下一个验证器，这样客户端可以编写
```
required.set_next(email_format).set_next(amount_range)
```
这样的链式代码。
模块划分：当验证逻辑复杂时，每个验证器单独一个模块（例如
```
validators/email_format.py
```
）；将协议和基础处理器放在
```
validators/base.py
```
中。
避免过度设计：如果只有一两个固定步骤，简单函数就足够了；当步骤较多或需要动态组合时，再使用CoR模式。
通用链场景：同样的模式也适用于非验证流水线（例如数据转换、数据增强、多步骤处理）——只需使用符合领域需求的上下文，以及能处理并传递的处理器即可。

Pipeline vs Chain of Responsibility

流水线 vs 责任链

Feature	Pipeline	Chain of Responsibility
Execution	Fixed, mandatory sequence	Conditional; handler decides whether to pass to the next
Flow	Linear, no branching	Allows flexible termination and branching
Termination	Runs to completion (barring errors)	Can be terminated early by a handler
Use cases	Data processing, parsing, ETL	Event handling, approval workflows, validation, message filtering

Use Pipeline when every stage must run in a fixed order (e.g. data transformation: parse → normalize → enrich → serialize). Use CoR when handlers can short-circuit or decide not to pass (e.g. validation, approval chains).

特性	流水线	责任链
执行方式	固定、强制的顺序	条件执行；处理器决定是否传递给下一个
流程	线性，无分支	支持灵活终止和分支
终止方式	除非出错，否则执行到完成	处理器可以提前终止流程
适用场景	数据处理、解析、ETL	事件处理、审批流程、验证、消息过滤

当每个阶段必须按固定顺序执行时，使用流水线（例如数据转换：解析 → 标准化 → 增强 → 序列化）。当处理器可以提前终止或决定不传递时，使用CoR（例如验证、审批链）。

Reference

参考资料

Chain of Responsibility — Refactoring.Guru: intent, problem/solution, structure, applicability, pros/cons, relations with Command/Decorator/Composite.

Chain of Responsibility — Refactoring.Guru：包含意图、问题/解决方案、结构、适用场景、优缺点，以及与命令/装饰器/组合模式的关系。