FastAPI + 设计模式 —— 从基础到实战
文档版本：1.0
目标读者：有 Python 基础，想系统理解设计模式在 FastAPI 中如何落地的后端开发者。
阅读方式：按章节顺序阅读，每章依赖前一章的知识。

第一章：背景与整体架构

1.1 什么是设计模式

定义：设计模式是针对软件设计中特定问题的可复用解决方案。它不是可以直接复制的代码，而是一种经验总结出的"模板"或"范式"。

设计模式按用途分为三大类：

分类 解决什么问题 常见模式
创建型（Creational） 对象如何创建 工厂模式、单例模式、建造者模式
结构型（Structural） 类和对象如何组合成更大的结构 适配器模式、装饰器模式、代理模式
行为型（Behavioral） 对象之间如何交互和分配职责 策略模式、观察者模式、责任链模式

为什么需要设计模式：

  • 应对变化：软件需求不断变化，设计模式将"变"的部分与"不变"的部分分离。
  • 解耦：减少模块间的直接依赖，使修改影响范围可控。
  • 复用经验：前人总结的成熟方案，避免重复踩坑。
  • 提高可读性：使用模式名称本身就是一种文档，团队成员能快速理解代码意图。

注意：设计模式不是银弹。过度设计或在不合适的场景使用，会增加不必要的复杂度。实践中遵循"三次原则"——同一类问题出现三次以上再考虑抽象。

1.2 什么是 FastAPI

定义：FastAPI 是一个现代 Python Web 框架，用于构建 RESTful API。

核心特性：

  • 异步支持：基于 Starlette（ASGI 框架），支持 async/await。
  • 自动生成文档：基于 OpenAPI 标准，自动生成 Swagger UI 和 ReDoc。
  • 依赖注入系统：通过 Depends 机制，将依赖（如数据库会话、服务对象）从业务逻辑中解耦。
  • 类型提示驱动：使用 Python 类型提示定义请求/响应模型，自动完成校验和序列化。

为什么选择 FastAPI：

  • 对设计模式的表达能力优于 Flask 等传统框架。
  • 依赖注入机制天然支持"面向接口编程"。
  • 本文讨论的所有设计模式都可以在 FastAPI 中落地。

1.3 本文的业务场景说明

为了具体展示设计模式的应用，本文统一使用支付渠道对接作为贯穿全文的业务场景。

为什么选择支付场景：

  • 支付渠道（支付宝、微信、PayPal 等）变化频繁，不同渠道的参数和逻辑差异大。
  • 新增或变更渠道是常见的产品需求，天然需要"对扩展开放，对修改关闭"。
  • 该场景足够简单（核心逻辑是 pay(amount)），便于聚焦设计模式本身，而非业务复杂性。

1.4 三层关系总览

graph TD
    subgraph 路由层
        API["/pay POST
接收请求 → 返回响应"]
    end
    subgraph 依赖注入层
        DI["Depends(get_payment_factory)
提供工厂实例"]
    end
    subgraph 工厂层
        Factory["PaymentFactory
根据 channel 创建具体产品"]
        Registry["_registry
渠道 → 类 映射表"]
    end
    subgraph 产品层
        Alipay["Alipay
支付宝支付逻辑"]
        WechatPay["WechatPay
微信支付逻辑"]
    end

    API --> DI --> Factory --> Registry --> Alipay & WechatPay

阅读说明：

  • 第一章是语法前提，熟悉 OOP 基础的可快速浏览。
  • 第二章是核心，重点理解工厂模式和单例模式的"是什么、为什么、怎么做"。
  • 第三章是将设计模式接入 FastAPI 框架。

第二章：Python OOP 基础（设计模式的语法前提）

本章为后续设计模式的实现提供语法铺垫。所有示例均围绕支付场景展开，方便与后续章节衔接。

2.1 什么是类，什么是实例

定义：

  • 类（Class）：一组对象的蓝图或模板，定义了数据（属性）和行为（方法）。
  • 实例（Instance）：根据类创建的具体对象，每个实例拥有独立的内存空间。

类属性 vs 实例属性：

  • 类属性：定义在类体中、方法之外。属于类本身，所有实例共享。
  • 实例属性：定义在 init 方法中，通过 self.xxx 赋值。属于具体实例，每个实例独有。
# 类定义（蓝图）
class Payment:
    # 类属性：所有实例共享
    currency: str = "CNY"

    def __init__(self, amount: float):
        # 实例属性：每个实例独有
        self.amount = amount


# 实例化
p1 = Payment(100.0)
p2 = Payment(200.0)

# 类属性通过类或实例访问
print(Payment.currency)   # CNY
print(p1.currency)        # CNY（实例无同名属性时向上查找类）

# 实例属性只能通过实例访问
print(p1.amount)          # 100.0
print(Payment.amount)     # AttributeError（类没有实例属性）

关键点：init 在实例化时执行，作用是将传入的参数挂载到当前实例上。它不是"构造函数"，而是"初始化方法"。真正的构造由 new 完成，本章第四节会涉及。

2.2 什么是继承

定义：继承是一种类之间的关系，子类（Subclass）从父类（Superclass）获得属性和方法，并可以添加或覆盖（Override）自己的实现。

super() 的作用：在子类中调用父类的方法。最常见的是在子类的 init 中调用父类的 init，确保父类中定义的实例属性被正确初始化。

class BasePayment:
    def __init__(self, amount: float):
        self.amount = amount

    def pay(self) -> str:
        return f"支付 {self.amount} 元"


class Alipay(BasePayment):
    def __init__(self, amount: float, app_id: str):
        # 调用父类 __init__，将 amount 挂到当前实例
        super().__init__(amount)
        self.app_id = app_id

    def pay(self) -> str:
        # 调用父类方法并扩展
        base = super().pay()
        return f"{base}，使用支付宝"


payment = Alipay(100.0, "app_123")
print(payment.pay())  # 支付 100.0 元，使用支付宝

两种调用父类方式的对比：

# 方式一：super() —— 推荐
super().__init__(amount)          # 自动传递 self

# 方式二：通过类名显式调用 —— 不推荐
BasePayment.__init__(self, amount)  # 必须手动传 self

为什么推荐 super()：

  • 不需要显式传递 self，减少错误。
  • 在多继承场景下，super() 遵循 MRO（方法解析顺序），比固定类名更安全。
  • 父类名称变更时，子类无需修改。
  • super() 不限于 init：可以调用父类的任何方法（super().pay()、super().validate() 等）。

2.3 什么是多继承与 Mixin

定义：

  • 多继承：一个子类可以同时继承多个父类。Python 支持多继承。
  • Mixin：一种特殊的多继承用法。Mixin 类提供可复用的方法，但不设计为独立实例化。它通常不包含 init 方法，只封装行为。

Mixin 规则：

  • 不写 init，避免构造方法传参混乱。
  • 只封装方法，假设 self 已经具备所需属性（由主类提供）。
  • 命名通常以 Mixin 结尾。
# 主基类：定义核心属性
class BasePayment:
    def __init__(self, amount: float):
        self.amount = amount


# Mixin 类：只封装方法，不写 __init__
class RetryMixin:
    def retry_pay(self, times: int = 3) -> str:
        for i in range(times):
            result = self.pay()  # 假设 self 有 pay() 方法
            if "成功" in result:
                return result
        return "支付失败：重试次数用尽"


# 主类：同时继承主基类和 Mixin
class Alipay(BasePayment, RetryMixin):
    def pay(self) -> str:
        return f"支付宝支付 {self.amount} 元"


payment = Alipay(100.0)
print(payment.retry_pay())  # 支付宝支付 100.0 元

何时使用 Mixin：

  • 需要为多个不相关的类添加相同的能力（如日志、重试、缓存）。
  • 这些能力不改变类的核心身份（Alipay 的本质是支付，不是"可重试支付"）。

2.4 什么是抽象方法

定义：抽象方法是在父类中声明但不实现的方法，由子类必须重写实现。@abstractmethod 是 Python 的 abc 模块提供的装饰器。

作用：

  • 定义"契约"：父类规定子类必须提供某个能力，但不关心具体实现。
  • 实例化时校验（Fail-Fast）：如果子类没有重写抽象方法，实例化时会立即报错，而非等到运行时才暴露。
from abc import ABC, abstractmethod

# 抽象基类：定义了契约
class Payment(ABC):
    @abstractmethod
    def pay(self, amount: float) -> str:
        """子类必须实现此方法"""
        pass


# 正确实现：可以实例化
class Alipay(Payment):
    def pay(self, amount: float) -> str:
        return f"支付宝支付 {amount} 元"


# 错误实现：没有重写 pay()
class WechatPay(Payment):
    pass


alipay = Alipay()  # 正常
# wechat = WechatPay()  # TypeError: Can't instantiate abstract class WechatPay without pay()

在工厂模式中的作用：

  • 工厂返回的是抽象类型 Payment，调用者只关心 payment.pay() 是否存在。
  • @abstractmethod 保证了所有具体支付类都实现了 pay()，工厂不会生产出"哑巴"产品。

第三章：工厂模式（Factory Pattern）

3.1 什么是工厂模式

定义：工厂模式是一种创建型设计模式，它将对象的创建逻辑封装到独立的工厂类或工厂方法中，调用方通过工厂获取对象，而无需直接 new 具体类。

分类：

  • 简单工厂（Simple Factory）：一个工厂类根据传入参数决定创建哪个具体产品。本文使用此模式。
  • 工厂方法（Factory Method）：定义一个创建对象的接口，由子类决定实例化哪个类。
  • 抽象工厂（Abstract Factory）：创建一系列相关或依赖的对象，不指定具体类。

为什么需要工厂模式：

  • 调用方不需要知道具体类的构造细节（如支付宝需要 app_id 和 private_key）。
  • 新增或修改具体类时，只需修改工厂，调用方代码不变。
  • 便于单元测试（可以注入 Mock 工厂）。

3.2 没有工厂的代码什么样

from typing import Literal

# 具体支付类
class Alipay:
    def __init__(self, app_id: str, private_key: str):
        self.app_id = app_id
        self.private_key = private_key

    def pay(self, amount: float) -> str:
        return f"支付宝支付 {amount} 元"


class WechatPay:
    def __init__(self, mch_id: str, api_key: str):
        self.mch_id = mch_id
        self.api_key = api_key

    def pay(self, amount: float) -> str:
        return f"微信支付 {amount} 元"


# 业务函数：直接依赖具体类
def process_payment(channel: str, amount: float) -> str:
    if channel == "alipay":
        payment = Alipay("app_123", "key_456")
    elif channel == "wechat":
        payment = WechatPay("mch_789", "key_012")
    else:
        raise ValueError(f"不支持: {channel}")
    return payment.pay(amount)


# 调用
print(process_payment("alipay", 100.0))

痛点：

  • 业务函数与具体类耦合（Alipay、WechatPay 写死在函数体）。
  • 新增渠道（如 PayPal）需要修改 process_payment（违反开闭原则）。
  • 测试 process_payment 时必须真正实例化支付宝/微信，需要真实的密钥配置。

3.3 使用简单工厂重构

# 抽象产品（可选，但推荐）
from abc import ABC, abstractmethod

class Payment(ABC):
    @abstractmethod
    def pay(self, amount: float) -> str:
        pass


# 具体产品
class Alipay(Payment):
    def __init__(self, app_id: str, private_key: str):
        self.app_id = app_id
        self.private_key = private_key

    def pay(self, amount: float) -> str:
        return f"支付宝支付 {amount} 元"


class WechatPay(Payment):
    def __init__(self, mch_id: str, api_key: str):
        self.mch_id = mch_id
        self.api_key = api_key

    def pay(self, amount: float) -> str:
        return f"微信支付 {amount} 元"


# 工厂类：封装创建逻辑
class PaymentFactory:
    @staticmethod
    def create(channel: str) -> Payment:
        if channel == "alipay":
            return Alipay("app_123", "private_key_456")
        elif channel == "wechat":
            return WechatPay("mch_789", "api_key_012")
        else:
            raise ValueError(f"不支持: {channel}")


# 业务函数：只依赖工厂，不依赖具体类
def process_payment(channel: str, amount: float) -> str:
    payment = PaymentFactory.create(channel)
    return payment.pay(amount)


# 调用
print(process_payment("alipay", 100.0))

收益：

  • process_payment 不再依赖 Alipay、WechatPay 具体类。
  • 新增渠道只修改 PaymentFactory.create，业务函数不变。
  • 测试时可以替换工厂返回 Mock 对象。

3.4 处理构造参数不一致的问题

不同支付渠道的构造参数不同：

  • 支付宝：app_id, private_key
  • 微信：mch_id, api_key

方案一：args, *kwargs（不推荐）

class PaymentFactory:
    @staticmethod
    def create(channel: str, *args, **kwargs) -> Payment:
        if channel == "alipay":
            return Alipay(*args, **kwargs)
        elif channel == "wechat":
            return WechatPay(*args, **kwargs)

问题：调用方必须知道每个渠道的参数顺序和个数，破坏封装。如果支付宝调整参数，所有调用方都要改。

方案二：Pydantic 参数对象（推荐）

from pydantic import BaseModel

# 统一参数对象（包含所有渠道的可能字段）
class CreatePaymentRequest(BaseModel):
    channel: str
    amount: float
    # 支付宝需要的
    app_id: str | None = None
    private_key: str | None = None
    # 微信需要的
    mch_id: str | None = None
    api_key: str | None = None


class PaymentFactory:
    @staticmethod
    def create(req: CreatePaymentRequest) -> Payment:
        if req.channel == "alipay":
            if not req.app_id or not req.private_key:
                raise ValueError("缺少支付宝参数")
            return Alipay(req.app_id, req.private_key)
        elif req.channel == "wechat":
            if not req.mch_id or not req.api_key:
                raise ValueError("缺少微信参数")
            return WechatPay(req.mch_id, req.api_key)
        else:
            raise ValueError(f"不支持: {req.channel}")


# 调用方
req = CreatePaymentRequest(
    channel="alipay",
    amount=100.0,
    app_id="app_123",
    private_key="key_456"
)
payment = PaymentFactory.create(req)
print(payment.pay(req.amount))

收益：

  • 调用方只需要构造一个 Pydantic 对象，参数由 FastAPI 自动解析。
  • 参数校验集中管理，工厂负责检查参数完整性。
  • 新增渠道只需在模型中增加字段，调用方代码不改。

方案三：工厂方法模式（消除 if-else）

from typing import Type, Dict

class Payment(ABC):
    @abstractmethod
    def pay(self, amount: float) -> str: pass

    @staticmethod
    @abstractmethod
    def from_request(req: CreatePaymentRequest) -> "Payment":
        pass


class Alipay(Payment):
    def __init__(self, app_id: str, private_key: str):
        self.app_id = app_id
        self.private_key = private_key

    def pay(self, amount: float) -> str:
        return f"支付宝支付 {amount} 元"

    @staticmethod
    def from_request(req: CreatePaymentRequest) -> "Payment":
        if not req.app_id or not req.private_key:
            raise ValueError("缺少支付宝参数")
        return Alipay(req.app_id, req.private_key)


class WechatPay(Payment):
    def __init__(self, mch_id: str, api_key: str):
        self.mch_id = mch_id
        self.api_key = api_key

    def pay(self, amount: float) -> str:
        return f"微信支付 {amount} 元"

    @staticmethod
    def from_request(req: CreatePaymentRequest) -> "Payment":
        if not req.mch_id or not req.api_key:
            raise ValueError("缺少微信参数")
        return WechatPay(req.mch_id, req.api_key)


class PaymentFactory:
    _registry: Dict[str, Type[Payment]] = {
        "alipay": Alipay,
        "wechat": WechatPay,
    }

    @staticmethod
    def create(req: CreatePaymentRequest) -> Payment:
        cls = PaymentFactory._registry.get(req.channel)
        if cls is None:
            raise ValueError(f"不支持: {req.channel}")
        # cls 是类引用（内存地址），不是实例
        # cls.from_request(req) 才真正创建实例
        return cls.from_request(req)

关键点：

  • cls = _registry.get("alipay") 返回的是类 Alipay 的引用（类似门牌号），不是实例。
  • cls.from_request(req) 调用类的静态方法，内部完成实例化。
  • 新增渠道：写一个新类 + 注册到 _registry，工厂代码本身不修改。

类图：

classDiagram
    class Payment {
        <>
        +pay(amount: float) str*
        +from_request(req) Payment*
    }
    class Alipay {
        -app_id: str
        -private_key: str
        +pay(amount: float) str
        +from_request(req) Payment
    }
    class WechatPay {
        -mch_id: str
        -api_key: str
        +pay(amount: float) str
        +from_request(req) Payment
    }
    class PaymentFactory {
        -_registry: Dict
        +create(req) Payment
    }
    Payment <|-- Alipay
    Payment <|-- WechatPay
    PaymentFactory ..> Payment : creates
    PaymentFactory ..> CreatePaymentRequest : uses

第四章：单例模式（Singleton Pattern）

4.1 什么是单例模式

定义：单例模式确保一个类有且只有一个实例，并提供一个全局访问点。

适用场景：

  • 数据库连接池（避免频繁创建和销毁连接）
  • 配置中心（全局统一配置）
  • 日志器（统一日志输出）
  • 缓存对象（如 Redis 客户端）

为什么需要单例：

  • 某些资源创建成本高，重复创建浪费内存和性能。
  • 某些状态必须全局一致（如配置项），多实例会导致数据不同步。

4.2 单例模式的标准实现

class DatabaseConnection:
    _instance = None      # 存储唯一实例的类属性
    _initialized = False  # 防止 __init__ 重复执行

    def __new__(cls):
        if cls._instance is None:
            cls._instance = super().__new__(cls)
        return cls._instance

    def __init__(self):
        if not DatabaseConnection._initialized:
            self.connection_string = "postgresql://localhost:5432/mydb"
            print("数据库连接初始化（仅一次）")
            DatabaseConnection._initialized = True

    def query(self, sql: str) -> str:
        return f"执行: {sql}"


# 验证
db1 = DatabaseConnection()
db2 = DatabaseConnection()
print(db1 is db2)          # True
print(id(db1) == id(db2))  # True

执行流程：

  1. 第一次调用 DatabaseConnection()：
  2. new 检查 _instance 为空 → 创建新实例 → 存入 _instance
  3. init 检查 _initialized 为 False → 执行初始化 → 设为 True
  4. 第二次调用：
  5. new 检查 _instance 非空 → 直接返回已有实例
  6. init 检查 _initialized 为 True → 跳过初始化

注意：init 每次都会被调用，所以必须通过标记防止重复初始化。

4.3 常见误区

# 普通多实例（不是单例）
db1 = DatabaseConnection()  # 实例 A
db2 = DatabaseConnection()  # 实例 B（不同内存地址）

# 如果不重写 __new__，每次都是新实例
print(db1 is db2)  # False

区分：

  • "只创建了一个实例"是偶然的（代码里只写了一次 DatabaseConnection()）。
  • 单例模式是强制性的，无论在代码中调用多少次，都返回同一个实例。

面试追问：单例模式在分布式系统中还成立吗？

不成立。单例模式保证的是"单个进程内"的唯一实例。
在分布式环境中（多个服务实例），每个进程各有自己的单例，需要借助 Redis 或 Zookeeper 等外部协调服务实现全局单例。

4.4 工厂模式 vs 单例模式对比

维度 工厂模式 单例模式
定义 封装对象的创建逻辑 确保类只有一个实例
解决什么问题 解耦调用方和具体类 全局统一资源、节省内存
实现方式 独立的 create 方法 重写 new，控制 _instance
每次调用的结果 通常返回新实例 始终返回同一个实例
适用场景 支付渠道创建、数据源切换 数据库连接池、配置中心

两者可以组合：工厂本身可以是单例，但工厂生产的产品通常是新实例。

class PaymentFactory:
    _instance = None
    def __new__(cls):
        if cls._instance is None:
            cls._instance = super().__new__(cls)
        return cls._instance

    def create(self, channel: str) -> Payment:
        # 每次返回新实例
        if channel == "alipay":
            return Alipay(...)

第五章：FastAPI 集成实战

5.1 什么是依赖注入

定义：依赖注入是一种设计原则，对象的依赖（即它需要的外部资源）由外部创建并"注入"到该对象中，而不是由对象自己创建。

FastAPI 的 Depends 机制：

  • 在路径操作函数（路由）的参数中使用 Depends，声明该参数由框架在运行时解析。
  • FastAPI 会执行 Depends 指定的函数，并将返回值注入到参数中。
  • 支持异步、支持依赖链（依赖可以依赖其他依赖）。

为什么需要 Depends：

  • 路由层不负责创建依赖对象，只声明"我需要什么"。
  • 依赖的生命周期由框架管理，便于测试时替换。
  • 代码更干净，职责边界清晰。

5.2 将工厂作为依赖注入

from fastapi import FastAPI, Depends, HTTPException
from pydantic import BaseModel

app = FastAPI()


# 请求模型
class PayRequest(BaseModel):
    channel: str
    amount: float
    app_id: str | None = None
    private_key: str | None = None
    mch_id: str | None = None
    api_key: str | None = None


# 响应模型
class PayResponse(BaseModel):
    result: str


# 依赖函数：返回工厂实例
def get_payment_factory() -> PaymentFactory:
    return PaymentFactory()


# 路由：通过 Depends 注入工厂
@app.post("/pay", response_model=PayResponse)
def pay(
    req: PayRequest,
    factory: PaymentFactory = Depends(get_payment_factory)
):
    try:
        # 将请求转换为创建参数（使用工厂方法模式）
        payment = factory.create(req)
        result = payment.pay(req.amount)
        return PayResponse(result=result)
    except ValueError as e:
        raise HTTPException(status_code=400, detail=str(e))

路由层的职责：

  • 接收 HTTP 请求 → 解析为 PayRequest 模型。
  • 声明需要 PaymentFactory（由 Depends 注入）。
  • 调用工厂创建支付对象 → 执行业务逻辑 → 返回响应。
  • 没有任何 if-else，没有任何具体类名（Alipay、WechatPay 不出现在路由中）。

5.3 将单例模式与 Depends 结合

在 FastAPI 中，通常不手动重写 new，而是利用框架的能力实现单例。

方案一：使用 @lru_cache

from functools import lru_cache

class DatabaseConnection:
    def __init__(self):
        self.connection_string = "postgresql://localhost:5432/mydb"
        print("数据库连接初始化")

    def query(self, sql: str) -> str:
        return f"执行: {sql}"


@lru_cache
def get_db() -> DatabaseConnection:
    # 此函数只会执行一次，后续调用返回缓存结果
    return DatabaseConnection()


@app.get("/users")
def get_users(db: DatabaseConnection = Depends(get_db)):
    return {"data": db.query("SELECT * FROM users")}

@lru_cache 的机制：首次调用 get_db() 时执行函数体并缓存返回值；后续调用直接返回缓存中的对象，实现了单例效果。

方案二：使用全局变量

_db_instance = None

def get_db() -> DatabaseConnection:
    global _db_instance
    if _db_instance is None:
        _db_instance = DatabaseConnection()
    return _db_instance

方案一更简洁，方案二更显式。两者均可。

5.4 完整路由代码（含所有模型）

from fastapi import FastAPI, Depends, HTTPException
from pydantic import BaseModel
from typing import Literal
from abc import ABC, abstractmethod
from functools import lru_cache

# ========== 1. 抽象产品 ==========
class Payment(ABC):
    @abstractmethod
    def pay(self, amount: float) -> str:
        pass

    @staticmethod
    @abstractmethod
    def from_request(req: "PayRequest") -> "Payment":
        pass


# ========== 2. 具体产品 ==========
class Alipay(Payment):
    def __init__(self, app_id: str, private_key: str):
        self.app_id = app_id
        self.private_key = private_key

    def pay(self, amount: float) -> str:
        return f"支付宝支付 {amount} 元"

    @staticmethod
    def from_request(req: "PayRequest") -> Payment:
        if not req.app_id or not req.private_key:
            raise ValueError("缺少支付宝参数")
        return Alipay(req.app_id, req.private_key)


class WechatPay(Payment):
    def __init__(self, mch_id: str, api_key: str):
        self.mch_id = mch_id
        self.api_key = api_key

    def pay(self, amount: float) -> str:
        return f"微信支付 {amount} 元"

    @staticmethod
    def from_request(req: "PayRequest") -> Payment:
        if not req.mch_id or not req.api_key:
            raise ValueError("缺少微信参数")
        return WechatPay(req.mch_id, req.api_key)


# ========== 3. 工厂 ==========
class PaymentFactory:
    _registry = {
        "alipay": Alipay,
        "wechat": WechatPay,
    }

    def create(self, req: "PayRequest") -> Payment:
        cls = self._registry.get(req.channel)
        if cls is None:
            raise ValueError(f"不支持: {req.channel}")
        return cls.from_request(req)


# ========== 4. FastAPI 应用 ==========
app = FastAPI()


class PayRequest(BaseModel):
    channel: Literal["alipay", "wechat"]
    amount: float
    app_id: str | None = None
    private_key: str | None = None
    mch_id: str | None = None
    api_key: str | None = None


class PayResponse(BaseModel):
    result: str
    trace_id: str | None = None


# 依赖注入：工厂本身作为单例
@lru_cache
def get_payment_factory() -> PaymentFactory:
    return PaymentFactory()


@app.post("/pay", response_model=PayResponse)
def pay(
    req: PayRequest,
    factory: PaymentFactory = Depends(get_payment_factory)
) -> PayResponse:
    try:
        payment = factory.create(req)
        result = payment.pay(req.amount)
        return PayResponse(result=result)
    except ValueError as e:
        raise HTTPException(status_code=400, detail=str(e))

第六章：完整流程串联

6.1 从 HTTP 请求到支付执行

6.2 各层的职责边界

层级 职责 不做什么
路由层（/pay） HTTP 请求/响应处理 不创建支付对象，不做业务逻辑
依赖注入层（Depends） 提供工厂实例 不创建具体支付类
工厂层（PaymentFactory） 根据 channel 创建具体支付对象 不执行业务逻辑
具体支付类（Alipay） 执行具体支付逻辑 不关心自己如何被创建

第七章：下一步学习计划

7.1 已覆盖的核心概念

分类 具体内容
Python OOP 基础 类与实例、类属性/实例属性、继承与 super、Mixin、抽象方法
创建型设计模式 工厂模式（简单工厂 + 工厂方法）、单例模式
FastAPI 集成 依赖注入 Depends、@lru_cache 实现单例、Pydantic 模型
代码组织 抽象产品 → 具体产品 → 工厂 → 路由分层

7.2 建议下一步学习的知识

分类 具体内容 与本文的关系
策略模式 将"算法"（如不同渠道的手续费计算）封装为可互换的对象 与工厂模式互补，工厂负责"创建"，策略负责"行为"
仓库模式 将数据访问逻辑封装到独立的 Repository 层 进一步解耦业务逻辑与数据库操作
观察者模式 通过事件驱动实现异步解耦（如支付成功发通知） 解决跨模块通信问题
中间件 FastAPI 的中间件机制（日志、限流、鉴权） 横切关注点与业务逻辑分离
BackgroundTasks FastAPI 的异步后台任务 支付成功后的异步处理（邮件通知、积分更新）
pytest + 依赖注入 利用依赖注入替换测试中的 Mock 对象 将本文的模式应用于单元测试
Docker 容器化应用交付 将代码固化为可移植的运行环境

7.3 学习路径建议

阶段一：巩固当前内容（1 周）

  • 手敲第三章所有代码，至少两遍。
  • 尝试在 _registry 中注册一个新的支付渠道（如 PayPal）。
  • 写单元测试覆盖 PaymentFactory.create() 的三种情况（alipay、wechat、不支持的渠道）。

阶段二：学习策略模式（1 周）

  • 理解策略模式的核心思想：封装算法族，使其可互换。
  • 将"渠道手续费计算"抽成独立策略，与工厂模式配合。
  • 实现一个"动态路由"：根据订单金额自动选择最优渠道。

阶段三：学习仓库模式（1 周）

  • 将支付记录存储到数据库。
  • 定义 PaymentRepository 接口，实现 PostgreSQL 版本。
  • 在 Alipay.pay() 中调用仓库保存记录。

阶段四：学习观察者模式 + 异步（1 周）

  • 支付成功后发布事件 PaymentSuccessEvent。
  • 编写监听器：发送邮件、记录审计日志、更新用户积分。
  • 使用 FastAPI 的 BackgroundTasks 实现异步执行。

阶段五：容器化交付（1 周）

  • 编写 Dockerfile 将 FastAPI 应用打包。
  • 使用 docker-compose 编排 FastAPI + PostgreSQL + Redis。
  • 配置环境变量注入工厂参数（如切换支付渠道的密钥）。

7.4 推荐资源

书籍：

  • 《设计模式：可复用面向对象软件的基础》（GoF 经典，建议阅读工厂模式、策略模式、观察者模式章节）
  • 《Python 设计模式》（适合 Python 开发者）

官方文档：

  • FastAPI 官方文档（重点阅读 Dependencies 章节）
  • Python abc 模块文档
  • Pydantic 官方文档

实操建议：

  • 不要一口气学完所有模式，每学一个模式就在自己的 Demo 项目中跑一遍。
  • 阅读开源项目源码（如 FastAPI 本身、SQLAlchemy），观察它们在哪些地方使用了设计模式。

附录：本文全部代码的目录结构

project/
├── main.py                 # FastAPI 入口
├── models.py               # Pydantic 请求/响应模型
├── dependencies.py         # 依赖注入函数（get_payment_factory、get_db）
├── payments/
│   ├── __init__.py
│   ├── base.py             # Payment 抽象类
│   ├── alipay.py           # Alipay 具体实现
│   ├── wechat.py           # WechatPay 具体实现
│   └── factory.py          # PaymentFactory
└── tests/
    ├── __init__.py
    ├── test_factory.py     # 工厂模式单元测试
    └── test_api.py         # FastAPI 集成测试