适配器模式:Python 中让不兼容接口和谐共舞的艺术

“优秀的程序员不是在制造轮子,而是在搭建桥梁。”


一、为什么你需要适配器?

每一位有实战经验的 Python 开发者,都曾遭遇过这样的困境:

你的系统已经稳定运行了一年,日志模块、数据库层、第三方 API 客户端各就各位。某天产品经理说:"我们需要接入新的支付平台。"你满怀信心地打开新平台的 SDK 文档,结果发现——它的接口设计风格和你现有的支付抽象层完全不同。旧接口是 pay(order_id, amount),新 SDK 却是 execute_transaction(payload: dict)

你有几个选择:

  • 重写现有代码:改动量大,引入新 bug 的风险极高
  • 到处写 if/else 判断:代码迅速腐烂,未来维护者(可能是你自己)会悄悄诅咒你
  • 用适配器模式优雅地解决:新旧接口互不干扰,测试照跑,逻辑清晰

聪明的选择显而易见。今天这篇文章,我们就来彻底拆解 Python 中的适配器模式(Adapter Pattern)——它是什么、为何有效、怎么实现,以及一个完整的真实第三方库适配案例。


二、适配器模式:一句话理解

适配器模式(Adapter Pattern)是 GoF 23 种经典设计模式之一,属于结构型模式

它的核心思想只有一句话

在不修改已有代码的前提下,用一个"翻译层"将不兼容的接口转换为客户端期望的接口。

生活中最直观的比喻是电源适配器:你的笔记本是美式两脚插头,欧洲的插座是圆形两孔,你不需要改变插座,也不需要换笔记本,只需要一个旅行转换头——这就是适配器。

在软件世界里,它的结构如下:

Client(客户端)
    │
    ▼
Target Interface(目标接口,客户端期望的)
    │
    ▼
Adapter(适配器,持有 Adaptee 的引用)
    │
    ▼
Adaptee(被适配者,第三方或旧有代码)

三、Python 基础实现:从最简单的例子开始

在深入真实案例之前,先用一个干净的例子建立直觉。

假设我们有一个旧的日志系统:

# 旧有接口:老日志模块(我们无法修改它)
class OldLogger:
    def write_log(self, level: str, message: str):
        print(f"[{level.upper()}] {message}")

但我们的新系统期望所有日志组件都遵循统一的接口:

# 目标接口:新系统期望的日志规范
from abc import ABC, abstractmethod

class Logger(ABC):
    @abstractmethod
    def info(self, message: str): pass

    @abstractmethod
    def error(self, message: str): pass

    @abstractmethod
    def debug(self, message: str): pass

二者接口完全不同。这时我们写一个适配器:

# 适配器:让 OldLogger 看起来像 Logger
class OldLoggerAdapter(Logger):
    def __init__(self, old_logger: OldLogger):
        self._adaptee = old_logger  # 持有被适配者的引用

    def info(self, message: str):
        self._adaptee.write_log("info", message)

    def error(self, message: str):
        self._adaptee.write_log("error", message)

    def debug(self, message: str):
        self._adaptee.write_log("debug", message)


# 使用方:完全感知不到 OldLogger 的存在
def process_order(logger: Logger, order_id: str):
    logger.info(f"开始处理订单: {order_id}")
    # ... 业务逻辑 ...
    logger.info(f"订单 {order_id} 处理完成")


# 组装
old_logger = OldLogger()
adapter = OldLoggerAdapter(old_logger)
process_order(adapter, "ORD-2024-001")

输出:

[INFO] 开始处理订单: ORD-2024-001
[INFO] 订单 ORD-2024-001 处理完成

process_order 函数只知道 Logger 接口,完全不关心底层是新系统还是旧系统。这就是适配器的魔力——解耦


四、真实案例:适配多个第三方支付 SDK

理论够了,让我们进入真实战场。这是工作中最常遇到的场景之一。

背景设定

你在开发一个电商平台,需要同时支持以下两个支付服务:

  • 支付宝 SDK(假设):采用面向过程的函数式调用风格
  • Stripe Python SDK(真实存在):面向对象,stripe.PaymentIntent.create()

将来还可能接入微信支付、PayPal 等。如果每次接if provider == ‘alipay’`,那维护成本会呈指数级上升。

第一步:定义统一的目标接口

# payment/base.py
from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import Optional


@dataclass
class PaymentResult:
    """统一的支付结果数据结构"""
    success: bool
    transaction_id: str
    amount: float
    currency: str
    error_message: Optional[str] = None


class PaymentGateway(ABC):
    """所有支付网关必须遵循的统一接口"""

    @abstractmethod
    def charge(self, amount: float, currency: str, token: str) -> PaymentResult:
        """发起支付"""
        pass

    @abstractmethod
    def refund(self, transaction_id: str, amount: float) -> PaymentResult:
        """发起退款"""
        pass

    @abstractmethod
    def query(self, transaction_id: str) -> PaymentResult:
        """查询交易状态"""
        pass

这是整个适配器体系的基石。所有适配器都必须实现这三个方法,业务代码永远只和 PaymentGateway 打交道。

第二步:模拟一个"支付宝风格"的 SDK(被适配者 A)

# third_party/alipay_sdk.py(模拟第三方库,无法修改)

import uuid

class AlipayClient:
    """支付宝 SDK,接口风格与我们的系统不兼容"""

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

    def unified_order(self, out_trade_no: str, total_fee: int, currency: str) -> dict:
        """
        发起支付。注意:
        - 金额单位是【分】(整数),而非元
        - 返回的是原始字典
        - 字段命名是下划线风格的中文业务语义
        """
        print(f"[AlipaySDK] 发起支付: 订单号={out_trade_no}, 金额={total_fee}分")
        return {
            "return_code": "SUCCESS",
            "trade_no": f"ALI{uuid.uuid4().hex[:16].upper()}",
            "out_trade_no": out_trade_no,
            "total_fee": total_fee,
        }

    def refund_apply(self, trade_no: str, refund_fee: int) -> dict:
        print(f"[AlipaySDK] 发起退款: 交易号={trade_no}, 退款金额={refund_fee}分")
        return {
            "return_code": "SUCCESS",
            "refund_id": f"RF{uuid.uuid4().hex[:12].upper()}",
        }

    def query_trade(self, trade_no: str) -> dict:
        return {
            "return_code": "SUCCESS",
            "trade_status": "TRADE_SUCCESS",
            "trade_no": trade_no,
            "total_fee": 9900,
        }

注意关键的不兼容点:金额单位是"分",而我们的统一接口用的是"元"(浮点数)。这类单位转换陷阱是实际开发中引发 Bug 最多的地方之一。

第三步:编写支付宝适配器

# payment/adapters/alipay_adapter.py

import uuid
from third_party.alipay_sdk import AlipayClient
from payment.base import PaymentGateway, PaymentResult


class AlipayAdapter(PaymentGateway):
    """
    将 AlipayClient 的接口适配为统一的 PaymentGateway 接口。
    核心职责:
    1. 参数转换(元 → 分,字段名映射)
    2. 返回值标准化(dict → PaymentResult)
    3. 异常统一处理
    """

    def __init__(self, app_id: str, private_key: str):
        self._client = AlipayClient(app_id, private_key)

    @staticmethod
    def _yuan_to_fen(yuan: float) -> int:
        """将元转换为分,并做精度处理防止浮点误差"""
        return round(yuan * 100)

    @staticmethod
    def _fen_to_yuan(fen: int) -> float:
        return fen / 100.0

    def charge(self, amount: float, currency: str, token: str) -> PaymentResult:
        out_trade_no = f"ORD{uuid.uuid4().hex[:12].upper()}"
        try:
            resp = self._client.unified_order(
                out_trade_no=out_trade_no,
                total_fee=self._yuan_to_fen(amount),  # 关键转换
                currency=currency,
            )
            if resp.get("return_code") == "SUCCESS":
                return PaymentResult(
                    success=True,
                    transaction_id=resp["trade_no"],
                    amount=amount,
                    currency=currency,
                )
            return PaymentResult(
                success=False,
                transaction_id="",
                amount=amount,
                currency=currency,
                error_message=resp.get("return_msg", "未知错误"),
            )
        except Exception as e:
            return PaymentResult(
                success=False, transaction_id="",
                amount=amount, currency=currency,
                error_message=str(e),
            )

    def refund(self, transaction_id: str, amount: float) -> PaymentResult:
        try:
            resp = self._client.refund_apply(
                trade_no=transaction_id,
                refund_fee=self._yuan_to_fen(amount),
            )
            success = resp.get("return_code") == "SUCCESS"
            return PaymentResult(
                success=success,
                transaction_id=resp.get("refund_id", ""),
                amount=amount,
                currency="CNY",
                error_message=None if success else "退款失败",
            )
        except Exception as e:
            return PaymentResult(
                success=False, transaction_id="",
                amount=amount, currency="CNY",
                error_message=str(e),
            )

    def query(self, transaction_id: str) -> PaymentResult:
        resp = self._client.query_trade(transaction_id)
        amount = self._fen_to_yuan(resp.get("total_fee", 0))
        return PaymentResult(
            success=resp.get("trade_status") == "TRADE_SUCCESS",
            transaction_id=transaction_id,
            amount=amount,
            currency="CNY",
        )

第四步:用工厂模式管理多个适配器

# payment/factory.py

from payment.base import PaymentGateway
from payment.adapters.alipay_adapter import AlipayAdapter
# from payment.adapters.stripe_adapter import StripeAdapter  # 未来扩展


class PaymentGatewayFactory:
    """
    工厂类:根据配置返回对应的支付网关适配器。
    业务代码只需调用这个工厂,完全无需关心底层 SDK。
    """

    _registry = {}

    @classmethod
    def register(cls, name: str, gateway_cls):
        cls._registry[name] = gateway_cls

    @classmethod
    def create(cls, provider: str, **kwargs) -> PaymentGateway:
        gateway_cls = cls._registry.get(provider)
        if not gateway_cls:
            raise ValueError(f"不支持的支付提供商: {provider}")
        return gateway_cls(**kwargs)


# 注册适配器
PaymentGatewayFactory.register("alipay", AlipayAdapter)
# PaymentGatewayFactory.register("stripe", StripeAdapter)  # 未来只需这一行

第五步:业务代码——简洁、稳定、优雅

# order_service.py

from payment.factory import PaymentGatewayFactory


def process_payment(provider: str, amount: float, currency: str, token: str):
    """
    业务层代码完全不依赖任何具体 SDK。
    无论底层是支付宝、Stripe 还是微信支付,这里一行不用改。
    """
    gateway = PaymentGatewayFactory.create(
        provider,
        app_id="YOUR_APP_ID",
        private_key="YOUR_PRIVATE_KEY",
    )

    print(f"\n{'='*40}")
    print(f"发起支付: provider={provider}, amount={amount}{currency}")
    result = gateway.charge(amount, currency, token)

    if result.success:
        print(f"✅ 支付成功! 交易号: {result.transaction_id}")
        # 退款演示
        refund_result = gateway.refund(result.transaction_id, amount)
        print(f"退款状态: {'✅ 成功' if refund_result.success else '❌ 失败'}")
    else:
        print(f"❌ 支付失败: {result.error_message}")

    return result


# 运行
process_payment("alipay", amount=99.0, currency="CNY", token="tok_test")

输出:

========================================
发起支付: provider=alipay, amount=99.0CNY
[AlipaySDK] 发起支付: 订单号=ORD..., 金额=9900分
✅ 支付成功! 交易号: ALI...
[AlipaySDK] 发起退款: 交易号=ALI..., 退款金额=9900分
退款状态: ✅ 成功

五、适配器模式的两种 Python 实现风格

除了上面基于类继承的对象适配器,Python 还可以用另一种更轻量的方式。

函数式适配器(适合简单场景):

def make_alipay_gateway(app_id: str, private_key: str) -> PaymentGateway:
    """用闭包构造一个轻量适配器,无需定义完整的类"""
    client = AlipayClient(app_id, private_key)

    class _Adapter(PaymentGateway):
        def charge(self, amount, currency, token):
            resp = client.unified_order(
                out_trade_no=uuid.uuid4().hex,
                total_fee=round(amount * 100),
                currency=currency,
            )
            return PaymentResult(
                success=resp["return_code"] == "SUCCESS",
                transaction_id=resp.get("trade_no", ""),
                amount=amount, currency=currency,
            )
        def refund(self, transaction_id, amount): ...
        def query(self, transaction_id): ...

    return _Adapter()

两种风格各有适用场景:当适配逻辑复杂、需要单元测试、需要复用时选类适配器;当逻辑简单、一次性使用时,函数式/闭包适配器更轻巧。


六、最佳实践与常见陷阱

✅ 应该做的:

  • 单一职责:适配器只做接口转换,不要在里面加入业务逻辑
  • 防御性编程:第三方 SDK 随时可能抛出奇怪的异常,适配器应该统一捕获并转换为标准错误
  • 单元测试适配器:用 unittest.mock.patch mock 掉第三方 SDK,专注测试转换逻辑
  • 记录转换规则:金额单位、时间格式、字段映射这类转换,务必写清楚注释
# 测试示例
from unittest.mock import patch, MagicMock

def test_alipay_adapter_charge_success():
    with patch('third_party.alipay_sdk.AlipayClient.unified_order') as mock_pay:
        mock_pay.return_value = {
            "return_code": "SUCCESS",
            "trade_no": "ALI_TEST_123",
        }
        adapter = AlipayAdapter("test_app", "test_key")
        result = adapter.charge(99.0, "CNY", "token")

        assert result.success is True
        assert result.transaction_id == "ALI_TEST_123"
        assert result.amount == 99.0
        # 验证单位转换是否正确
        mock_pay.assert_called_once()
        call_kwargs = mock_pay.call_args[1]
        assert call_kwargs["total_fee"] == 9900  # 99元 → 9900分

❌ 常见陷阱:

  • 适配器做了太多事:如果你在适配器里写了缓存、重试、业务校验,那它已经不是适配器了,需要拆分
  • 忘记处理异常:第三方库的异常泄漏到业务层,会让调用方困惑
  • 适配器嵌套适配器:这通常意味着架构设计出了问题,该重新审视接口定义

七、总结:桥梁的价值

适配器模式教会我们一件事:好的代码不是控制一切,而是隔离变化

第三方库会升级、会被替换、会停止维护——这些都是你无法控制的。但你可以控制的是:在你的系统边界处竖立一道清晰的接口墙,让外部的变化止步于适配器,内部的业务逻辑岁月静好。

这正是高级工程师和初级工程师的分水岭之一:初级工程师问"这个功能怎么实现",高级工程师问"这个变化如何被隔离"。

适配器模式,就是回答后一个问题的最优解之一。


你在项目中有没有遇到过需要适配不同第三方接口的场景?你是如何解决的? 欢迎在评论区聊聊你的思路和踩过的坑,一起把这个话题的实践价值挖得更深。


附录:参考资料

Logo

Agent 垂直技术社区,欢迎活跃、内容共建。

更多推荐