适配器模式

适配器模式（Adapter Pattern）将一个类的接口转换成客户希望的另一个接口，使得原本由于接口不兼容而不能一起工作的那些类可以一起工作。

问题定义

场景1：第三方API集成

# ❌ 问题：接口不兼容
# 我们的支付接口
class PaymentProcessor:
def process_payment(self, amount):
"""我们的标准接口"""
print(f"处理支付: ${amount}")
# 第三方支付系统A
class PayPalAPI:
def make_payment(self, currency, amount):
"""PayPal的接口 - 不同参数顺序"""
print(f"PayPal支付: {currency} {amount}")
# 第三方支付系统B
class StripeAPI:
def charge(self, payment_data):
"""Stripe的接口 - 完全不同"""
print(f"Stripe支付: {payment_data}")
# 问题：无法统一使用
def process_order(payment_system, amount):
payment_system.process_payment(amount)  # PayPal和Stripe都没有这个方法！

场景2：旧代码复用

# 旧系统 - 接口不兼容
class LegacyPrinter:
def print_old(self, text):
print(f"[旧打印机] {text}")
# 新系统期望的接口
class NewPrinter:
def print(self, document):
print(f"[新打印机] {document}")
# 问题：无法直接使用旧打印机
def print_document(printer: NewPrinter, text):
printer.print(text)  # LegacyPrinter没有print方法

解决方案

适配器模式通过在中间加一个"适配器"层，转换接口调用。

标准实现

对象适配器

# 目标接口
class PaymentTarget:
"""支付接口 - 客户端期望的接口"""
def pay(self, amount):
"""标准支付方法"""
raise NotImplementedError
# 被适配者 - 第三方API
class PayPal:
"""PayPal支付API"""
def make_payment(self, currency, amount):
print(f"PayPal支付: {currency} {amount}")
return f"P-{amount}"
class Stripe:
"""Stripe支付API"""
def charge(self, payment_data):
print(f"Stripe支付: {payment_data}")
return f"S-{payment_data['amount']}"
# 适配器
class PayPalAdapter(PaymentTarget):
"""PayPal适配器"""
def __init__(self, paypal):
self.paypal = paypal
def pay(self, amount):
# 调用被适配者的方法，转换接口
return self.paypal.make_payment("USD", amount)
class StripeAdapter(PaymentTarget):
"""Stripe适配器"""
def __init__(self, stripe):
self.stripe = stripe
def pay(self, amount):
# 调用被适配者的方法，转换接口
payment_data = {"amount": amount, "currency": "USD"}
return self.stripe.charge(payment_data)
# 客户端使用
def process_payment(payment_system: PaymentTarget, amount):
"""处理支付 - 客户端只依赖PaymentTarget"""
transaction_id = payment_system.pay(amount)
print(f"交易ID: {transaction_id}")
# 使用适配器
paypal_adapter = PayPalAdapter(PayPal())
stripe_adapter = StripeAdapter(Stripe())
print("使用PayPal:")
process_payment(paypal_adapter, 100)
print("\n使用Stripe:")
process_payment(stripe_adapter, 100)

类适配器

Python使用多继承实现类适配器（较少使用）：

class PaymentTarget:
def pay(self, amount):
raise NotImplementedError
class PayPal:
def make_payment(self, currency, amount):
print(f"PayPal支付: {currency} {amount}")
# 类适配器 - 继承目标接口和被适配者
class PayPalAdapterClass(PaymentTarget, PayPal):
def pay(self, amount):
return self.make_payment("USD", amount)
# 使用
adapter = PayPalAdapterClass()
adapter.pay(100)  # PayPal支付: USD 100

实战应用

应用1：日志系统适配

# 目标接口
class Logger:
"""日志接口"""
def log(self, level, message):
"""标准日志方法"""
pass
# 第三方日志系统A
class ThirdPartyLoggerA:
def write_log(self, severity, text):
print(f"[LoggerA] {severity}: {text}")
# 第三方日志系统B
class ThirdPartyLoggerB:
def record(self, msg_type, content, timestamp):
print(f"[LoggerB] {timestamp} [{msg_type}] {content}")
# 适配器
import datetime
class LoggerAAdapter(Logger):
"""日志系统A适配器"""
def __init__(self, logger_a):
self.logger_a = logger_a
def log(self, level, message):
# 转换级别
severity_map = {
"INFO": "INFO",
"WARNING": "WARN",
"ERROR": "ERROR"
}
severity = severity_map.get(level, "INFO")
self.logger_a.write_log(severity, message)
class LoggerBAdapter(Logger):
"""日志系统B适配器"""
def __init__(self, logger_b):
self.logger_b = logger_b
def log(self, level, message):
# 转换接口
timestamp = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
self.logger_b.record(level, message, timestamp)
# 使用
def application_workflow(logger: Logger):
"""应用工作流"""
logger.log("INFO", "应用启动")
logger.log("WARNING", "配置文件缺失")
logger.log("ERROR", "数据库连接失败")
print("使用日志系统A:")
adapter_a = LoggerAAdapter(ThirdPartyLoggerA())
application_workflow(adapter_a)
print("\n使用日志系统B:")
adapter_b = LoggerBAdapter(ThirdPartyLoggerB())
application_workflow(adapter_b)

应用2：数据格式转换

# 目标接口
class DataProcessor:
"""数据处理器接口"""
def process(self, data):
"""处理数据"""
pass
# JSON数据源
class JSONDataSource:
def get_json(self):
return """
{
"name": "张三",
"age": 30,
"city": "北京"
}
"""
# XML数据源
class XMLDataSource:
def get_xml(self):
return """
<person>
<name>李四</name>
<age>25</age>
<city>上海</city>
</person>
"""
# CSV数据源
class CSVDataSource:
def get_csv(self):
return "name,age,city\n王五,28,广州"
# 适配器
import json
class JSONAdapter(DataProcessor):
"""JSON适配器"""
def __init__(self, json_source):
self.json_source = json_source
def process(self, data):
# 解析JSON
json_str = self.json_source.get_json()
parsed = json.loads(json_str)
print(f"处理JSON数据: {parsed}")
class XMLAdapter(DataProcessor):
"""XML适配器"""
def __init__(self, xml_source):
self.xml_source = xml_source
def process(self, data):
# 解析XML（简化版）
xml_str = self.xml_source.get_xml()
print(f"解析XML: {xml_str}")
class CSVAdapter(DataProcessor):
"""CSV适配器"""
def __init__(self, csv_source):
self.csv_source = csv_source
def process(self, data):
# 解析CSV
csv_str = self.csv_source.get_csv()
lines = csv_str.strip().split('\n')
headers = lines[0].split(',')
values = lines[1].split(',')
data_dict = dict(zip(headers, values))
print(f"处理CSV数据: {data_dict}")
# 使用
def load_data(adapter: DataProcessor):
"""加载数据"""
adapter.process("data")
print("加载JSON数据:")
load_data(JSONAdapter(JSONDataSource()))
print("\n加载XML数据:")
load_data(XMLAdapter(XMLDataSource()))
print("\n加载CSV数据:")
load_data(CSVAdapter(CSVDataSource()))

应用3：数据库驱动适配

# 目标接口
class DatabaseDriver:
"""数据库驱动接口"""
def connect(self, host, port, database):
pass
def execute(self, query):
pass
def disconnect(self):
pass
# 模拟的数据库驱动
class MySQLDriver:
def open_connection(self, server, port_num, db_name):
print(f"MySQL连接: {server}:{port_num}/{db_name}")
def run_query(self, sql):
print(f"MySQL查询: {sql}")
def close_connection(self):
print("MySQL连接关闭")
class PostgreSQLDriver:
def init_db(self, host, port, db):
print(f"PostgreSQL初始化: {host}:{port}/{db}")
def exec_sql(self, statement):
print(f"PostgreSQL执行: {statement}")
def terminate(self):
print("PostgreSQL终止连接")
# 适配器
class MySQLAdapter(DatabaseDriver):
"""MySQL适配器"""
def __init__(self, driver):
self.driver = driver
def connect(self, host, port, database):
self.driver.open_connection(host, port, database)
def execute(self, query):
self.driver.run_query(query)
def disconnect(self):
self.driver.close_connection()
class PostgreSQLAdapter(DatabaseDriver):
"""PostgreSQL适配器"""
def __init__(self, driver):
self.driver = driver
def connect(self, host, port, database):
self.driver.init_db(host, port, database)
def execute(self, query):
self.driver.exec_sql(query)
def disconnect(self):
self.driver.terminate()
# 使用
class Application:
"""应用程序"""
def __init__(self, db_driver: DatabaseDriver):
self.db = db_driver
def run(self):
self.db.connect("localhost", 3306, "mydb")
self.db.execute("SELECT * FROM users")
self.db.disconnect()
# 使用不同数据库
print("使用MySQL:")
app1 = Application(MySQLAdapter(MySQLDriver()))
app1.run()
print("\n使用PostgreSQL:")
app2 = Application(PostgreSQLAdapter(PostgreSQLDriver()))
app2.run()

双向适配器

双向适配器可以在两个方向上转换接口：

# 系统A的接口
class SystemA:
def method_a(self):
print("系统A的方法")
# 系统B的接口
class SystemB:
def method_b(self):
print("系统B的方法")
# 双向适配器
class TwoWayAdapter(SystemA, SystemB):
def __init__(self, system_a=None, system_b=None):
self.system_a = system_a
self.system_b = system_b
def method_a(self):
if self.system_b:
# 从系统B的接口转换为系统A
print("转换中...")
return self.system_b.method_b()
else:
super().method_a()
def method_b(self):
if self.system_a:
# 从系统A的接口转换为系统B
print("转换中...")
return self.system_a.method_a()
else:
super().method_b()
# 使用
a = SystemA()
b = SystemB()
# 创建双向适配器
adapter = TwoWayAdapter(system_a=a, system_b=b)
# 从系统A的角度使用
adapter.method_a()  # 调用系统B的方法
print()
# 从系统B的角度使用
adapter.method_b()  # 调用系统A的方法

默认适配器

提供默认实现，减少适配器的工作：

from abc import ABC, abstractmethod
class Logger(ABC):
"""日志接口"""
@abstractmethod
def log(self, message):
pass
def info(self, message):
"""默认实现 - 可以被子类覆盖"""
self.log(f"INFO: {message}")
def warning(self, message):
"""默认实现"""
self.log(f"WARNING: {message}")
def error(self, message):
"""默认实现"""
self.log(f"ERROR: {message}")
class ConsoleLogger(Logger):
"""控制台日志适配器 - 只需实现log方法"""
def log(self, message):
print(message)
# 使用
logger = ConsoleLogger()
logger.info("应用启动")
logger.warning("配置文件缺失")
logger.error("数据库连接失败")

适配器 vs 装饰器 vs 代理

优缺点

✅ 优点

❌ 缺点

本章要点

• ✅ 适配器模式转换接口，使不兼容的类一起工作
• ✅ 对象适配器：组合，更灵活
• ✅ 类适配器：继承，简单但较少用
• ✅ 双向适配器支持双向转换
• ✅ 适用于集成第三方API、旧代码复用
• ✅ 与装饰器、代理模式目的不同

下一步：装饰器模式 🚀