第 10 章 · 装饰器（Decorator）

一句话：把每个附加功能写成一层包装，包装和原对象接口相同，要哪几层、按什么顺序，在创建对象的地方一层套一层拼起来。 解决的痛：每种功能组合写一个子类，功能每多一个，子类数就翻一倍多；同一段压缩、加密代码抄在好几个类里，抄的时候顺序写反也没人发现。

10.1 一个场景

你在一家做 SaaS 报表系统的公司写后端。每个企业客户（下文叫租户）每天导出几十份订单报表，系统要把这些文件存起来，之后随时能下载。负责存取文件的是一个存储对象，只有两个方法：save(name, data) 把一段字节存成一个文件，load(name) 把它读回来。现在只有一种实现 DiskStorage，写本地磁盘；本章用一个字典代替磁盘，文件名是键，存下来的字节是值。

产品提了两个附加功能：

压缩：报表是 CSV，一行行的日期、单号、状态大量重复，压缩一下能省很多磁盘。用 Python 标准库的 zlib，TS 里用 Node 自带的 node:zlib。
加密：有些租户的合同要求文件在磁盘上必须是加密的。本章用“SHA-256 生成密钥流再逐字节异或”模拟加密，它和真正的加密一样会把数据变成看上去随机的字节；真实项目要用 cryptography 这类库里的 AES-GCM 等正经算法。

两个租户的要求不一样：租户 A 要压缩加加密；租户 B 上传的本来就是压缩包，只要加密。

两个功能叠在一起时顺序有讲究：先压缩，再加密。压缩靠的是找重复，加密后的数据看上去是随机字节，没有重复可找，这时再压缩基本压不小。读的时候反过来：先解密，再解压。

10.2 先按直觉写

存储已经有一个 DiskStorage 类，加功能最顺手的办法是继承它：压缩一个子类，加密一个子类，两样都要的再来一个子类。每个子类覆盖 save、load，处理完再调父类：

import hashlib
import zlib

KEY = b"tenant-key"


def xor_cipher(data: bytes) -> bytes:
    # 模拟加密：用 SHA-256 生成一串密钥流，再和数据逐字节异或；加密和解密是同一个操作。
    # 只为演示，真实项目用 cryptography 这类库里的 AES-GCM 等正经算法。
    out = bytearray()
    for i, byte in enumerate(data):
        if i % 32 == 0:  # 每 32 字节换一块密钥：SHA-256(密钥 + 块序号)
            block = hashlib.sha256(KEY + (i // 32).to_bytes(4, "big")).digest()
        out.append(byte ^ block[i % 32])
    return bytes(out)


class DiskStorage:  # 本地磁盘（用字典模拟）：文件名 → 存下来的字节
    def __init__(self) -> None:
        self.files: dict[str, bytes] = {}

    def save(self, name: str, data: bytes) -> None:
        self.files[name] = data

    def load(self, name: str) -> bytes:
        return self.files[name]


class CompressedStorage(DiskStorage):  # 压缩
    def save(self, name: str, data: bytes) -> None:
        super().save(name, zlib.compress(data))

    def load(self, name: str) -> bytes:
        return zlib.decompress(super().load(name))


class EncryptedStorage(DiskStorage):  # 加密
    def save(self, name: str, data: bytes) -> None:
        super().save(name, xor_cipher(data))

    def load(self, name: str) -> bytes:
        return xor_cipher(super().load(name))


class CompressedEncryptedStorage(DiskStorage):  # 压缩 + 加密：两段代码从上面两个类抄过来
    def save(self, name: str, data: bytes) -> None:
        super().save(name, xor_cipher(zlib.compress(data)))  # 先压缩，再加密

    def load(self, name: str) -> bytes:
        return zlib.decompress(xor_cipher(super().load(name)))  # 先解密，再解压


# 一份 400 行的订单报表，重复很多，压缩效果明显
REPORT = "".join(f"2025-01-{i % 28 + 1:02d},order-{1000 + i},shipped,{i * 37 % 500}.00\n" for i in range(400)).encode()

TENANT_STORAGE: dict[str, type[DiskStorage]] = {"A": CompressedEncryptedStorage, "B": EncryptedStorage}

if __name__ == "__main__":
    for tenant, storage_class in TENANT_STORAGE.items():
        storage = storage_class()
        storage.save("report.csv", REPORT)
        same = "一致" if storage.load("report.csv") == REPORT else "不一致"
        stored = len(storage.files["report.csv"])
        print(f"租户 {tenant}（{storage_class.__name__}）：{len(REPORT)} 字节 → 磁盘上 {stored} 字节，读回{same}")

# 输出：
# 租户 A（CompressedEncryptedStorage）：14714 字节 → 磁盘上 2368 字节，读回一致
# 租户 B（EncryptedStorage）：14714 字节 → 磁盘上 14714 字节，读回一致

TS 版结构一样，save、load 收发的是 Uint8Array：

import { createHash } from "node:crypto";
import { deflateSync, inflateSync } from "node:zlib";

const KEY = new TextEncoder().encode("tenant-key");

function xorCipher(data: Uint8Array): Uint8Array {
  // 模拟加密：用 SHA-256 生成一串密钥流，再和数据逐字节异或；加密和解密是同一个操作。
  // 只为演示，真实项目用 node:crypto 的 AES-GCM 等正经算法。
  const out = new Uint8Array(data.length);
  let block = new Uint8Array(0);
  for (let i = 0; i < data.length; i++) {
    if (i % 32 === 0) { // 每 32 字节换一块密钥：SHA-256(密钥 + 块序号)
      const index = Buffer.alloc(4);
      index.writeUInt32BE(i / 32);
      block = createHash("sha256").update(KEY).update(index).digest();
    }
    out[i] = data[i] ^ block[i % 32];
  }
  return out;
}

class DiskStorage { // 本地磁盘（用 Map 模拟）：文件名 → 存下来的字节
  readonly files = new Map<string, Uint8Array>();

  save(name: string, data: Uint8Array): void {
    this.files.set(name, data);
  }

  load(name: string): Uint8Array {
    const data = this.files.get(name);
    if (data === undefined) throw new Error(`没有这个文件：${name}`);
    return data;
  }
}

class CompressedStorage extends DiskStorage { // 压缩
  override save(name: string, data: Uint8Array): void {
    super.save(name, deflateSync(data));
  }

  override load(name: string): Uint8Array {
    return inflateSync(super.load(name));
  }
}

class EncryptedStorage extends DiskStorage { // 加密
  override save(name: string, data: Uint8Array): void {
    super.save(name, xorCipher(data));
  }

  override load(name: string): Uint8Array {
    return xorCipher(super.load(name));
  }
}

class CompressedEncryptedStorage extends DiskStorage { // 压缩 + 加密：两段代码从上面两个类抄过来
  override save(name: string, data: Uint8Array): void {
    super.save(name, xorCipher(deflateSync(data))); // 先压缩，再加密
  }

  override load(name: string): Uint8Array {
    return inflateSync(xorCipher(super.load(name))); // 先解密，再解压
  }
}

// 一份 400 行的订单报表，重复很多，压缩效果明显
const REPORT = new TextEncoder().encode(Array.from({ length: 400 }, (_, i) =>
  `2025-01-${String((i % 28) + 1).padStart(2, "0")},order-${1000 + i},shipped,${(i * 37) % 500}.00\n`).join(""));

const sameBytes = (a: Uint8Array, b: Uint8Array): boolean => a.length === b.length && a.every((x, i) => x === b[i]);

const TENANT_STORAGE: Record<string, new () => DiskStorage> = { A: CompressedEncryptedStorage, B: EncryptedStorage };

for (const [tenant, StorageClass] of Object.entries(TENANT_STORAGE)) {
  const storage = new StorageClass();
  storage.save("report.csv", REPORT);
  const same = sameBytes(storage.load("report.csv"), REPORT) ? "一致" : "不一致";
  const stored = storage.files.get("report.csv")?.length;
  console.log(`租户 ${tenant}（${StorageClass.name}）：${REPORT.length} 字节 → 磁盘上 ${stored} 字节，读回${same}`);
}

// 输出：
// 租户 A（CompressedEncryptedStorage）：14714 字节 → 磁盘上 2368 字节，读回一致
// 租户 B（EncryptedStorage）：14714 字节 → 磁盘上 14714 字节，读回一致

三个子类各管一种组合，TENANT_STORAGE 把租户对到类上。每个类的 save 和 load 互为逆操作，存进去、读出来内容一致，租户 A 的 14714 字节报表落到磁盘上只有 2368 字节。两种功能、两个租户时，这样写清楚直接，大多数人第一版都这么写。

10.3 需求变了

报表下载量上来以后，同一份报表常常被反复下载，每次都要读盘、解密、解压。新需求是：加一层缓存，最近读过的文件放在内存里，再读就不走磁盘；新租户 C 要缓存、压缩、加密三样都开，缓存放在最外层。别的租户以后也可能单独开缓存。

要改的地方有四处：

新增 CachedStorage(DiskStorage)：缓存用的字典、读的时候先查缓存、存的时候让旧缓存作废，十来行。这一处任何写法都躲不掉。
缓存要和已有的每种组合再组合一遍：新增 CachedCompressedStorage、CachedEncryptedStorage、CachedCompressedEncryptedStorage。第 1 处的缓存逻辑抄进这三个类，压缩、加密的调用也在里面再写一遍。三个功能的组合是 2³ − 1 = 7 个类，这次要新写 4 个；下一个功能（比如审计日志）来了就是 15 个。
压缩的那两行现在住在 4 个类里，加密也是 4 个。想把压缩级别从默认的 6 调到 9，要打开 4 个类各改一处，漏改一个，开了那种组合的租户就还是旧级别。
TENANT_STORAGE 加一行 "C": CachedCompressedEncryptedStorage。和第 1 处一样，任何写法都要在某处登记租户用什么。

Python 可以用多重继承，让方法解析顺序（Method Resolution Order，MRO）把几个父类的 save 串起来，但顺序就藏在基类列表的先后里；TS 的类只能继承一个类，这条路走不通。所以本书两种语言都按复制写。

事故出在第 2 处。CachedCompressedEncryptedStorage 是照着 EncryptedStorage 改出来的，加压缩时把 zlib.compress 套在了外面：

class CachedCompressedEncryptedStorage(DiskStorage):  # 缓存、压缩、加密的代码各抄一份
    def __init__(self) -> None:
        super().__init__()
        self.cache: dict[str, bytes] = {}

    def save(self, name: str, data: bytes) -> None:
        self.cache.pop(name, None)
        super().save(name, zlib.compress(xor_cipher(data)))  # 顺序反了：先加密，后压缩

    def load(self, name: str) -> bytes:
        if name not in self.cache:
            self.cache[name] = xor_cipher(zlib.decompress(super().load(name)))  # 读的顺序也跟着反了
        return self.cache[name]

load 和 save 恰好互为逆操作，存进去读出来内容一致，测试全过，上线了。一个月后运维发现租户 C 的磁盘占用不对劲：那份 14714 字节的报表存下来是 14725 字节，比原文还多 11 个字节；正确顺序只要 2368 字节，同样一份文件占的磁盘成了原来的 6 倍多。加密后的字节压不动，zlib 只是在外面加了一圈格式开销。10.4 的代码把两种顺序都存了一次，输出里能看到这两个数。这么久没人发现，是因为这个错不影响任何功能，只影响磁盘账单。

问题的形状是：每加一个功能，类的数量翻一倍再多一个；每段功能代码散在一半的类里，叠加的顺序在每个组合类里各写一遍，写反了也照样能跑。

继承写法：缓存加进来以后的子类和四处改动

图里标红的是这次要新写的 4 个类和要加一行的 TENANT_STORAGE，①–④ 对应上面四处改动；边框加粗的 CachedCompressedEncryptedStorage 就是顺序写反的那个。

10.4 用装饰器模式重写

换个思路：压缩不是“一种存储”，而是“给任意一个存储加上压缩”。那就让压缩层自己也是一个 Storage，同时手里拿着另一个 Storage：save 时先压缩，再交给手里那个；load 时先从手里那个读，再解压。加密、缓存同理。手里拿的可以是 DiskStorage，也可以是另一层包装，于是几层可以随便套：

import hashlib
import zlib
from collections.abc import Callable
from typing import Protocol

KEY = b"tenant-key"


def xor_cipher(data: bytes) -> bytes:  # 和 10.2 相同：模拟加密，真实项目用 AES-GCM 等正经算法
    out = bytearray()
    for i, byte in enumerate(data):
        if i % 32 == 0:
            block = hashlib.sha256(KEY + (i // 32).to_bytes(4, "big")).digest()
        out.append(byte ^ block[i % 32])
    return bytes(out)


class Storage(Protocol):  # 所有存储和装饰器都实现这两个方法
    def save(self, name: str, data: bytes) -> None: ...
    def load(self, name: str) -> bytes: ...


class DiskStorage:  # 被装饰的对象：本地磁盘（用字典模拟），和 10.2 相同
    def __init__(self) -> None:
        self.files: dict[str, bytes] = {}

    def save(self, name: str, data: bytes) -> None:
        self.files[name] = data

    def load(self, name: str) -> bytes:
        return self.files[name]


class StorageDecorator:  # 装饰器基类：持有一个 Storage，默认原样转发
    def __init__(self, inner: Storage) -> None:
        self.inner = inner

    def save(self, name: str, data: bytes) -> None:
        self.inner.save(name, data)

    def load(self, name: str) -> bytes:
        return self.inner.load(name)


class Compressed(StorageDecorator):  # 存之前压缩，读出来解压
    def save(self, name: str, data: bytes) -> None:
        self.inner.save(name, zlib.compress(data))

    def load(self, name: str) -> bytes:
        return zlib.decompress(self.inner.load(name))


class Encrypted(StorageDecorator):  # 存之前加密，读出来解密
    def save(self, name: str, data: bytes) -> None:
        self.inner.save(name, xor_cipher(data))

    def load(self, name: str) -> bytes:
        return xor_cipher(self.inner.load(name))


class Cached(StorageDecorator):  # 新加的缓存：就这一个类
    def __init__(self, inner: Storage) -> None:
        super().__init__(inner)
        self.cache: dict[str, bytes] = {}

    def save(self, name: str, data: bytes) -> None:
        self.cache.pop(name, None)  # 文件变了，旧缓存作废
        self.inner.save(name, data)

    def load(self, name: str) -> bytes:
        if name in self.cache:
            print(f"[缓存] 命中 {name}，不读磁盘")
            return self.cache[name]
        data = self.inner.load(name)
        self.cache[name] = data
        return data


# 组合和顺序都写在这里：缓存在最外层；Compressed 包在 Encrypted 外面，所以存的时候先压缩、后加密
TENANT_STORAGE: dict[str, Callable[[Storage], Storage]] = {
    "A": lambda disk: Compressed(Encrypted(disk)),
    "B": lambda disk: Encrypted(disk),
    "C": lambda disk: Cached(Compressed(Encrypted(disk))),
}

REPORT = "".join(f"2025-01-{i % 28 + 1:02d},order-{1000 + i},shipped,{i * 37 % 500}.00\n" for i in range(400)).encode()

if __name__ == "__main__":
    storages: dict[str, Storage] = {}
    for tenant, build in TENANT_STORAGE.items():
        disk = DiskStorage()
        storages[tenant] = storage = build(disk)
        storage.save("report.csv", REPORT)
        same = "一致" if storage.load("report.csv") == REPORT else "不一致"
        print(f"租户 {tenant}：{len(REPORT)} 字节 → 磁盘上 {len(disk.files['report.csv'])} 字节，读回{same}")
    storages["C"].load("report.csv")  # 租户 C 再读一次

    disk = DiskStorage()
    Encrypted(Compressed(disk)).save("report.csv", REPORT)  # 把两层写反：先加密，后压缩
    print(f"顺序写反：{len(REPORT)} 字节 → 磁盘上 {len(disk.files['report.csv'])} 字节")

# 输出：
# 租户 A：14714 字节 → 磁盘上 2368 字节，读回一致
# 租户 B：14714 字节 → 磁盘上 14714 字节，读回一致
# 租户 C：14714 字节 → 磁盘上 2368 字节，读回一致
# [缓存] 命中 report.csv，不读磁盘
# 顺序写反：14714 字节 → 磁盘上 14725 字节

TS 版的 StorageDecorator 用构造参数属性 protected readonly inner 持有下一层：

import { createHash } from "node:crypto";
import { deflateSync, inflateSync } from "node:zlib";

const KEY = new TextEncoder().encode("tenant-key");

function xorCipher(data: Uint8Array): Uint8Array { // 和 10.2 相同：模拟加密，真实项目用 AES-GCM 等正经算法
  const out = new Uint8Array(data.length);
  let block = new Uint8Array(0);
  for (let i = 0; i < data.length; i++) {
    if (i % 32 === 0) {
      const index = Buffer.alloc(4);
      index.writeUInt32BE(i / 32);
      block = createHash("sha256").update(KEY).update(index).digest();
    }
    out[i] = data[i] ^ block[i % 32];
  }
  return out;
}

interface Storage { // 所有存储和装饰器都实现这两个方法
  save(name: string, data: Uint8Array): void;
  load(name: string): Uint8Array;
}

class DiskStorage implements Storage { // 被装饰的对象：本地磁盘（用 Map 模拟），和 10.2 相同
  readonly files = new Map<string, Uint8Array>();

  save(name: string, data: Uint8Array): void {
    this.files.set(name, data);
  }

  load(name: string): Uint8Array {
    const data = this.files.get(name);
    if (data === undefined) throw new Error(`没有这个文件：${name}`);
    return data;
  }
}

class StorageDecorator implements Storage { // 装饰器基类：持有一个 Storage，默认原样转发
  constructor(protected readonly inner: Storage) {}

  save(name: string, data: Uint8Array): void {
    this.inner.save(name, data);
  }

  load(name: string): Uint8Array {
    return this.inner.load(name);
  }
}

class Compressed extends StorageDecorator { // 存之前压缩，读出来解压
  override save(name: string, data: Uint8Array): void {
    this.inner.save(name, deflateSync(data));
  }

  override load(name: string): Uint8Array {
    return inflateSync(this.inner.load(name));
  }
}

class Encrypted extends StorageDecorator { // 存之前加密，读出来解密
  override save(name: string, data: Uint8Array): void {
    this.inner.save(name, xorCipher(data));
  }

  override load(name: string): Uint8Array {
    return xorCipher(this.inner.load(name));
  }
}

class Cached extends StorageDecorator { // 新加的缓存：就这一个类
  private readonly cache = new Map<string, Uint8Array>();

  override save(name: string, data: Uint8Array): void {
    this.cache.delete(name); // 文件变了，旧缓存作废
    this.inner.save(name, data);
  }

  override load(name: string): Uint8Array {
    const hit = this.cache.get(name);
    if (hit !== undefined) {
      console.log(`[缓存] 命中 ${name}，不读磁盘`);
      return hit;
    }
    const data = this.inner.load(name);
    this.cache.set(name, data);
    return data;
  }
}

// 组合和顺序都写在这里：缓存在最外层；Compressed 包在 Encrypted 外面，所以存的时候先压缩、后加密
const TENANT_STORAGE: Record<string, (disk: Storage) => Storage> = {
  A: (disk) => new Compressed(new Encrypted(disk)),
  B: (disk) => new Encrypted(disk),
  C: (disk) => new Cached(new Compressed(new Encrypted(disk))),
};

const REPORT = new TextEncoder().encode(Array.from({ length: 400 }, (_, i) =>
  `2025-01-${String((i % 28) + 1).padStart(2, "0")},order-${1000 + i},shipped,${(i * 37) % 500}.00\n`).join(""));

const sameBytes = (a: Uint8Array, b: Uint8Array): boolean => a.length === b.length && a.every((x, i) => x === b[i]);

const storages: Record<string, Storage> = {};
for (const [tenant, build] of Object.entries(TENANT_STORAGE)) {
  const disk = new DiskStorage();
  const storage = (storages[tenant] = build(disk));
  storage.save("report.csv", REPORT);
  const same = sameBytes(storage.load("report.csv"), REPORT) ? "一致" : "不一致";
  console.log(`租户 ${tenant}：${REPORT.length} 字节 → 磁盘上 ${disk.files.get("report.csv")?.length} 字节，读回${same}`);
}
storages["C"].load("report.csv"); // 租户 C 再读一次

const disk = new DiskStorage();
new Encrypted(new Compressed(disk)).save("report.csv", REPORT); // 把两层写反：先加密，后压缩
console.log(`顺序写反：${REPORT.length} 字节 → 磁盘上 ${disk.files.get("report.csv")?.length} 字节`);

// 输出：
// 租户 A：14714 字节 → 磁盘上 2368 字节，读回一致
// 租户 B：14714 字节 → 磁盘上 14714 字节，读回一致
// 租户 C：14714 字节 → 磁盘上 2368 字节，读回一致
// [缓存] 命中 report.csv，不读磁盘
// 顺序写反：14714 字节 → 磁盘上 14725 字节

装饰器版：三个装饰器都实现 Storage，又都持有一个 Storage

看标红的 Cached：加缓存就是这一个类。它和 Compressed、Encrypted 一样继承 StorageDecorator，对外是 Storage，对内持有一个 Storage。

租户 C 存一次、读一次，调用穿过三层的顺序

看顺序图：save 从最外层往里走，Compressed 在 Encrypted 外面，所以先压缩、后加密；load 从最里层往外回，先解密、后解压，最后 Cached 把结果留一份。哪层在外面，哪层的事就在存的时候先做、读的时候后做。

逐条对照 10.3 的四处改动：

第 1 处，缓存逻辑：还是要写，十来行一行不少，现在写在 Cached 一个类里。
第 2 处，组合类：一个都不用写。租户 C 的组合就是 TENANT_STORAGE 里的一行 Cached(Compressed(Encrypted(disk)))。三个功能是三个装饰器，再来审计日志就是第四个，不会再翻倍。
第 3 处，改压缩级别：压缩只写在 Compressed 里，改一处，所有开了压缩的租户一起生效。
第 4 处，登记租户：TENANT_STORAGE 照样要加一行，装饰器省不掉这一行。不同的是，这一行现在还承担了顺序：叠加顺序从各个组合类里搬到了这里。

要说清楚的是，顺序事故并没有被模式消灭。Encrypted(Compressed(disk)) 一样能创建、能存能读，类型检查也不拦，代码最后一行就是这么写的，打印出 14725 字节。装饰器的改进在于：顺序原来散在每个组合类的 save、load 里，现在集中到了 TENANT_STORAGE 这几行，一眼能看全，能在代码评审里对着规则查。所以规则一定要写下来：像代码里那样在组装处写一句注释“缓存在最外层，压缩包在加密外面”，再补一个测试，断言一份重复很多的报表存下来比原文小。

代价也有：一个租户的存储是三四个对象套在一起，调试时要一层层点开 inner 才看到 DiskStorage；外层对象拿不到 files 这种只有底层才有的属性，所以示例代码要单独留着 disk 变量去查磁盘上的大小。

10.5 模式定义

GoF 对装饰器的定义是：动态地给一个对象添加一些额外的职责。就增加功能来说，装饰器模式相比生成子类更为灵活。

“动态”指的是功能在运行时、由创建对象的代码加上去，而不是在写类的时候用继承定死。同一个 DiskStorage 类，给租户 A 包两层，给租户 C 包三层，都不用新写类。“更灵活”说的就是 10.3 那笔账：n 个功能用子类要 2ⁿ − 1 个组合类，用装饰器是 n 个类，组合交给调用处。

装饰器能这样叠，靠的是两点：每个装饰器实现的接口和被装饰的对象相同，所以外面的人分不出包了几层；每个装饰器持有的也是这个接口，所以它包的可以是真对象，也可以是另一个装饰器。

装饰器的结构：Decorator 既实现 Component，又持有一个 Component

对照 10.4 的案例图：Storage 是 Component，DiskStorage 是 ConcreteComponent，Compressed、Encrypted、Cached 是具体装饰器。

角色	在本章案例里是谁	职责
组件（Component）	`Storage`	被装饰对象和装饰器共同的接口，这里是 `save`、`load`
具体组件（ConcreteComponent）	`DiskStorage`	真正干活的原对象，不知道自己被包了几层
装饰器（Decorator）	`StorageDecorator`	实现 Component，持有一个 Component，默认原样转发
具体装饰器（ConcreteDecorator）	`Compressed`、`Encrypted`、`Cached`	在转发前后加上自己的功能

常见场景：

I/O 流。缓冲、压缩各是一层，包在二进制文件对象外面，对外还是二进制文件，10.7 例一就是这个。
HTTP 客户端。重试、超时、日志、加鉴权头，各包一层在发请求的函数外面，见 10.7 例二。
UI 组件加边框、滚动条。GoF 书里的原始例子就是给一个文本视图包上边框和滚动条，视图本身不用知道。
权限与审计。给一个仓储对象包一层 Audited，每次读写记下谁在什么时候动了什么；再包一层权限检查，原来的仓储代码不改。

10.6 更简洁的写法

Python 程序员听到“装饰器”，先想到的往往是函数上面那行 @。被装饰的东西如果只是一个函数，而不是一个有好几个方法的对象，就用不着接口、基类和一个个类：函数本身可以当参数传，一个“收一个函数、返回一个包了一层的新函数”的函数就是装饰器。

import functools
from collections.abc import Callable

Fetch = Callable[[str], str]


def log_calls(fn: Fetch) -> Fetch:  # 函数装饰器：收一个函数，返回一个包了一层的新函数
    @functools.wraps(fn)  # 让新函数沿用原函数的 __name__ 和文档，调试时看得出是谁
    def wrapper(month: str) -> str:
        print(f"[日志] 参数 {month}")
        result = fn(month)
        print(f"[日志] 返回：{result}")
        return result
    return wrapper


def retry(fn: Fetch) -> Fetch:
    @functools.wraps(fn)
    def wrapper(month: str) -> str:
        for attempt in range(1, 4):
            try:
                return fn(month)
            except ConnectionError as error:
                print(f"[重试] 第 {attempt} 次失败：{error}")
        raise ConnectionError("重试 3 次仍然失败")
    return wrapper


calls = 0


@log_calls  # 两行 @ 等价于在函数定义之后写：fetch_report = log_calls(retry(fetch_report))
@retry
def fetch_report(month: str) -> str:  # 模拟从报表服务取数据：前两次超时
    global calls
    calls += 1
    if calls < 3:
        raise ConnectionError("报表服务超时")
    return f"{month} 月报，共 400 行"


if __name__ == "__main__":
    fetch_report("2025-01")

# 输出：
# [日志] 参数 2025-01
# [重试] 第 1 次失败：报表服务超时
# [重试] 第 2 次失败：报表服务超时
# [日志] 返回：2025-01 月报，共 400 行

TS 里对应的是高阶函数（higher-order function），名字习惯叫 withXxx：

type Fetch = (month: string) => string;

function withLogging(fn: Fetch): Fetch { // 高阶函数：收一个函数，返回一个包了一层的新函数
  return (month) => {
    console.log(`[日志] 参数 ${month}`);
    const result = fn(month);
    console.log(`[日志] 返回：${result}`);
    return result;
  };
}

function withRetry(fn: Fetch): Fetch {
  return (month) => {
    for (let attempt = 1; attempt <= 3; attempt++) {
      try {
        return fn(month);
      } catch (error) {
        console.log(`[重试] 第 ${attempt} 次失败：${(error as Error).message}`);
      }
    }
    throw new Error("重试 3 次仍然失败");
  };
}

let calls = 0;

function rawFetchReport(month: string): string { // 模拟从报表服务取数据：前两次超时
  calls += 1;
  if (calls < 3) throw new Error("报表服务超时");
  return `${month} 月报，共 400 行`;
}

// 和 Python 的两行 @ 是同一件事；TS 的 @ 装饰器语法只能用在类和类成员上，本章不用
const fetchReport = withLogging(withRetry(rawFetchReport));

fetchReport("2025-01");

// 输出：
// [日志] 参数 2025-01
// [重试] 第 1 次失败：报表服务超时
// [重试] 第 2 次失败：报表服务超时
// [日志] 返回：2025-01 月报，共 400 行

三件事要分清：

@ 只是语法糖（syntactic sugar）。@log_calls 写在 def fetch_report 上面，等价于定义完以后执行一句 fetch_report = log_calls(fetch_report)；两行 @ 从下往上套，等价于 log_calls(retry(fetch_report))。TS 那一行 withLogging(withRetry(rawFetchReport)) 就是把这句话直接写出来。
装饰的东西不同，组合的时机不同。装饰器模式装饰的是对象，一个对象可以有好几个方法（save 和 load 要一起包），组合在运行时决定，同一个类可以给 A 租户包两层、给 C 租户包三层。@ 装饰的是一个函数，在函数定义时就套好了，之后所有调用 fetch_report 的地方拿到的都是包好的版本；想按租户换一种包法，得回到手写 log_calls(retry(f)) 的写法。
两者同名，是因为做的是同一件事：包一层，不改原物。retry 里的 wrapper 就相当于一个只有一个方法的具体装饰器，fn 就是它持有的组件。

语言替你省掉的是 Storage 接口、StorageDecorator 基类和每个装饰器的类定义：只有一个方法时，“实现同一个接口”就是“签名相同的函数”，“持有下一层”就是闭包里的 fn。TS 也有 @decorator 语法（TS 5 实现了 ECMAScript 的装饰器提案），但它只能用在类和类的成员上，不能直接装饰一个普通函数，本章不用。

10.7 真实系统里的例子

例一：Python io 模块的流（Python）

open("report.csv", "w", encoding="utf-8") 返回的对象其实是三层：最里面是 io.FileIO，只会把字节交给操作系统，每次 write 都是一次系统调用；外面包一层 io.BufferedWriter，把零碎的小写入攒成一大块再往下写；最外面是 io.TextIOWrapper，把字符串按编码变成字节。可以自己验证：f.buffer 是那个 BufferedWriter，f.buffer.raw 是 FileIO。这三层里只有中间那层是装饰器：BufferedWriter 对外还是一个二进制文件，只是多了缓冲；TextIOWrapper 把“写字节”变成了“写字符串”，接口换了，它是第 9 章讲的适配器。标准库的 gzip.GzipFile(fileobj=...) 和 BufferedWriter 一样是装饰器：它包在任何一个二进制文件对象外面，自己也是二进制文件，写进去的是明文，交给下一层的是 gzip 数据。下面用三个小类模拟“裸文件 → 缓冲 → gzip”这一串：

import gzip
import zlib


class Stream:  # 三层对外一样：大致相当于 io 里可写的二进制文件对象；计数只为演示每层收到了什么
    def __init__(self) -> None:
        self.calls, self.nonempty, self.total = 0, 0, 0

    def write(self, data: bytes) -> None:
        self.calls += 1
        self.nonempty += 1 if data else 0
        self.total += len(data)

    def flush(self) -> None: ...

    def report(self) -> str:
        return f"{type(self).__name__}：收到 {self.calls} 次 write（非空 {self.nonempty} 次），共 {self.total} 字节"


class RawFile(Stream):  # 大致相当于 io.FileIO：每次 write 都是一次系统调用
    def __init__(self) -> None:
        super().__init__()
        self.content = b""

    def write(self, data: bytes) -> None:
        super().write(data)
        self.content += data


class Buffered(Stream):  # 大致相当于 io.BufferedWriter：攒够一块再往下写（真实默认 8 KB）
    def __init__(self, raw: Stream, size: int = 1024) -> None:
        super().__init__()
        self.raw, self.size, self.buf = raw, size, b""

    def write(self, data: bytes) -> None:
        super().write(data)
        self.buf += data
        if len(self.buf) >= self.size:
            self.flush()

    def flush(self) -> None:
        if self.buf:
            self.raw.write(self.buf)
            self.buf = b""


class GzipLayer(Stream):  # 大致相当于 gzip.GzipFile(fileobj=..., mode="wb")：收明文，往下交 gzip 数据
    def __init__(self, fileobj: Stream) -> None:
        super().__init__()
        self.fileobj, self.compressor = fileobj, zlib.compressobj(wbits=31)  # 31 表示 gzip 格式

    def write(self, data: bytes) -> None:
        super().write(data)
        self.fileobj.write(self.compressor.compress(data))  # zlib 自己也攒数据，多数时候返回空

    def flush(self) -> None:  # 真实的 GzipFile 在 close() 时写出结尾，这里合并成 flush
        self.fileobj.write(self.compressor.flush())
        self.fileobj.flush()


if __name__ == "__main__":
    raw = RawFile()
    buffered = Buffered(raw)
    f = GzipLayer(buffered)  # 一层包一层，最外层是 gzip
    for i in range(300):
        f.write(f"order-{1000 + i},shipped\n".encode())
    f.flush()
    for layer in (f, buffered, raw):
        print(layer.report())
    print(f"用标准库 gzip 解压：{len(gzip.decompress(raw.content))} 字节")

# 输出：
# GzipLayer：收到 300 次 write（非空 300 次），共 5700 字节
# Buffered：收到 301 次 write（非空 2 次），共 625 字节
# RawFile：收到 1 次 write（非空 1 次），共 625 字节
# 用标准库 gzip 解压：5700 字节

RawFile 对应本章的 DiskStorage，Buffered、GzipLayer 对应 Cached、Compressed：每层都是 Stream，都有 write 和 flush，每层只管自己那件事，再交给手里的下一层。看输出：外面 300 次 write、5700 字节明文进了 GzipLayer；zlib 的压缩器自己也会攒数据，只有第一次吐出 10 字节的 gzip 头，其余 299 次都返回空，flush 时才吐出剩下的 615 字节，所以 Buffered 收到 301 次 write，非空的只有 2 次；这两段加起来 625 字节，不到 1024 字节的块大小，Buffered 先攒着，flush 时合成一次交给 RawFile。落下去的是标准库 gzip.decompress 认得的 gzip 数据。真实的 BufferedWriter 还处理读和 seek，GzipFile 还会在头里写文件名和时间戳，这里都省略了。

例二：给 fetch 包上重试、超时、日志（TypeScript）

前端和 Node 项目里，很少有人直接裸用 fetch。ky、wretch 这类库在 fetch 外面提供重试、超时、钩子等功能，大致就是包在 fetch 外面的一层层逻辑；不用库的项目也常自己写 withRetry(fetch)、withTimeout(fetch)、withLogging(fetch) 这样的包装函数，每个都收一个“像 fetch 的函数”，返回一个签名相同的函数。简化版：

type Fetch = (url: string) => Promise<string>; // 真实的 fetch 是 (input, init) => Promise<Response>，这里简化

let attempts = 0;
const fakeFetch: Fetch = async (url) => { // 模拟网络：前两次连接被重置
  attempts += 1;
  if (attempts < 3) throw new Error("ECONNRESET");
  return `{"url":"${url}","orders":3}`;
};

function withTimeout(fetchFn: Fetch, ms: number): Fetch { // 超过 ms 毫秒就当失败
  return async (url) => {
    let timer: ReturnType<typeof setTimeout> | undefined;
    const timeout = new Promise<never>((_, reject) => {
      timer = setTimeout(() => reject(new Error(`超时 ${ms}ms`)), ms);
    });
    try {
      return await Promise.race([fetchFn(url), timeout]);
    } finally {
      clearTimeout(timer);
    }
  };
}

function withRetry(fetchFn: Fetch, times: number): Fetch { // 失败了再试，最多 times 次
  return async (url) => {
    for (let attempt = 1; ; attempt++) {
      try {
        return await fetchFn(url);
      } catch (error) {
        console.log(`[重试] 第 ${attempt} 次失败：${(error as Error).message}`);
        if (attempt >= times) throw error;
      }
    }
  };
}

function withLogging(fetchFn: Fetch): Fetch { // 请求前后各记一条日志
  return async (url) => {
    console.log(`[日志] GET ${url}`);
    const body = await fetchFn(url);
    console.log(`[日志] 完成 ${url}，响应 ${body.length} 字节`);
    return body;
  };
}

// 每次尝试各自限时 500ms，最多试 3 次，整个过程只记一条请求日志
const api = withLogging(withRetry(withTimeout(fakeFetch, 500), 3));

void api("/api/orders").then((body) => console.log(`拿到：${body}，网络层一共被调了 ${attempts} 次`));

// 输出：
// [日志] GET /api/orders
// [重试] 第 1 次失败：ECONNRESET
// [重试] 第 2 次失败：ECONNRESET
// [日志] 完成 /api/orders，响应 32 字节
// 拿到：{"url":"/api/orders","orders":3}，网络层一共被调了 3 次

Fetch 类型对应 Storage 接口，fakeFetch 对应 DiskStorage，三个 withXxx 是三个具体装饰器，只不过被装饰的是函数，所以写成了 10.6 那种高阶函数。顺序在这里同样要紧：withTimeout 在 withRetry 里面，每次尝试各自限时 500ms；要是写成 withTimeout(withRetry(...))，500ms 就成了三次尝试加起来的总时限。withLogging 在最外层，所以三次尝试只记一条请求日志。

10.8 什么时候别用

只有一种附加功能，而且永远不和别的叠加。比如只有导出文件需要压缩，也没有第二个功能要来。写一个 CompressedStorage 子类，或者在 save 里加一个 if compress:，读的人一眼看懂；为它建接口、基类和装饰器，要多读两个类才知道发生了什么。
包到看不出对象是谁。一个存储对象如果套了七层（缓存、压缩、加密、审计、限流、重试、指标），断点停下来时变量面板里是 inner.inner.inner...，要连点七次才看到底下是磁盘还是对象存储；异常栈里七层 save 挨在一起，分不清是哪层抛的。层数多时至少给每层写一个能打印出整串结构的 __repr__，或者把组装收进一个有名字的函数。
装饰器之间有顺序依赖，却没人写下来。10.3 的事故在装饰器写法里一样会发生，只要有人把 Compressed(Encrypted(disk)) 写成 Encrypted(Compressed(disk))，程序照跑，磁盘占用变成 6 倍多。顺序规则只活在写代码那个人的脑子里时，装饰器的灵活就是在给下一个人埋坑：把规则写在组装处，并用测试守住。

10.9 容易混淆的模式

模式	意图	区别
装饰器（Decorator）	不改原对象，给它叠加功能	去掉它，功能少了一样（不压缩了、不缓存了）
代理（Proxy）	用一个替身控制对原对象的访问	结构几乎相同，都是同接口、持有原对象；去掉它，功能没少，只是访问没人把关了（谁都能读、什么时候都能连），见第 12 章
适配器（Adapter）	让接口对不上的类按调用方要的接口工作	适配器换接口，装饰器不换接口，见第 9 章
Python 的 `@` 语法	定义函数时把它换成包了一层的版本	`@deco` 就是 `f = deco(f)`；包的是一个函数、定义时固定，装饰器模式包的是有多个方法的对象、运行时组合，见 10.6

判别装饰器和代理最实用的一句：“去掉它，功能少了”是装饰器；“去掉它，功能一样，只是访问没人把关了”是代理。和责任链的分别：装饰器每一层都一定往下传，责任链的一层可以就此停下，见第 23 章。

10.10 练一练

有一个文本来源接口 TextSource，只有一个方法 read()，返回一段多行文字。已有一个最简单的实现 StringSource，构造时传入一段文字，read() 原样返回。

请用装饰器实现：

写装饰器基类 TextDecorator，持有一个 TextSource，默认原样转发。
写三个具体装饰器：UpperCase 把文字转成大写；Trimmed 去掉每行首尾的空白，并丢掉空行；Numbered 在每行前面加上行号，格式是 1: 内容。
用 " save report \n\n load report" 做原文，按两种顺序组合：Numbered(Trimmed(UpperCase(source))) 和 Trimmed(Numbered(UpperCase(source)))，打印两种结果。Python 和 TS 各写一版。

提示：先自己预测两种顺序的输出，再运行对照。想一想第二种顺序里空行为什么没被丢掉、行首的空白为什么还在：Trimmed 拿到的是什么？

10.11 小结

装饰器把每个附加功能写成一层包装：和原对象实现同一个接口，再持有一个同接口的对象，做完自己的事就转给它。
n 个功能用子类组合要 2ⁿ − 1 个类，用装饰器是 n 个类；加功能是加一个类，要哪几层由创建对象的地方决定。
叠加顺序从各个组合类里搬到了组装处，但没有消失：压缩和加密写反照样能跑。顺序规则要写在组装处，并用测试守住。
Python 的 @deco 是 f = deco(f) 的语法糖，装饰的是函数、定义时固定；装饰器模式装饰的是对象、运行时组合。两者同名，因为都是包一层不改原物。
和代理结构几乎一样：去掉后功能少了的是装饰器，去掉后访问没人把关了的是代理。