《接口变更管理：电商平台API一升级，你的ERP代码就要改？适配器模式破局》（附Python源码）
先拍结论：
电商平台API升级不可怕，可怕的是你的业务代码里到处散落着response["data"]["items"][0]["price"]。适配器模式的核心不是“加一层”，而是把平台差异锁死在适配器内部——平台改API，只改适配器；业务逻辑一行不动。二手ERP尤其需要这个：闲鱼、Mercari、Back Market、eBay四个平台，每个每年至少2-3次breaking change，没有适配器就是改不完的bug。
一、平台API变更的四种痛法
类型 | 例子 | 不改适配器的后果 |
|---|---|---|
字段改名 | item_price → price_amount | 业务代码 order["item_price"] 全挂 |
数据结构重组 | 平铺变嵌套： price → pricing.total.amount | 取值链断掉 |
分页参数变更 | page → cursor | 全量翻页逻辑重写 |
认证方式变更 | Token → OAuth2 | 每个请求都要改header构造 |
适配器模式的承诺：以上四种变更，只改适配器，不改一行业务代码。
二、适配器模式核心：三层隔离
┌─────────────────────────────────────────────────────────────┐ │ 业务逻辑层 │ │ 订单处理 │ 库存同步 │ 定价引擎 │ 报表统计 │ │ (只认 InternalOrder / InternalProduct) │ └──────────────────────────┬──────────────────────────────────┘ │ 统一接口 ┌──────────────────────────▼──────────────────────────────────┐ │ 适配器抽象层 │ │ PlatformAdapter (ABC) │ │ + fetch_order(order_id) -> InternalOrder │ │ + sync_listing(product) -> ListingResult │ │ + update_inventory(sku, qty) -> bool │ └──────────┬──────────────┬──────────────┬────────────────────┘ │ │ │ ┌──────────▼──┐ ┌────────▼──────┐ ┌───▼─────────────┐ │ XianyuAdapter│ │ MercariAdapter│ │ BackMarketAdapter│ │ v2.3 │ │ v1.8 │ │ v3.1 │ │ (闲鱼API 2026)│ │ (Mercari API) │ │ (BM API v3) │ └──────────────┘ └───────────────┘ └──────────────────┘
规则：
业务层 只知道
InternalOrder/InternalProduct，不知道tradeDTO/itemPayload/listingResponse适配器层 只负责翻译，不做业务判断
平台API升级 → 只换对应适配器的实现类，业务代码零改动
三、完整源码：适配器模式 + 版本兼容 + 优雅降级
# adapter_pattern.py
"""
电商平台适配器模式
- PlatformAdapter: 抽象基类（统一接口）
- XianyuAdapter / MercariAdapter / BackMarketAdapter: 具体实现
- AdapterRegistry: 按平台+版本注册
- VersionMigration: API版本迁移工具
- FallbackChain: 降级策略（新版本挂了切旧版本）
"""
import abc
import time
import json
from typing import Dict, List, Optional, Tuple
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
# ==================== 统一数据模型（业务层只认这些） ====================
@dataclass
class InternalOrder:
order_id: str
platform: str
sku: str
quantity: int
price: float
currency: str
status: str
buyer_name: str = ""
shipping_address: str = ""
created_at: datetime = field(default_factory=datetime.now)
platform_raw: dict = field(default_factory=dict) # 原始数据存档
@dataclass
class InternalProduct:
sku: str
title: str
description: str
price: float
currency: str
stock: int
images: List[str] = field(default_factory=list)
attributes: dict = field(default_factory=dict)
@dataclass
class ListingResult:
success: bool
platform_listing_id: str = ""
errors: List[str] = field(default_factory=list)
# ==================== 适配器抽象基类 ====================
class PlatformAdapter(abc.ABC):
"""所有平台适配器必须实现的接口"""
@abc.abstractmethod
def fetch_order(self, order_id: str) -> InternalOrder:
...
@abc.abstractmethod
def sync_listing(self, product: InternalProduct) -> ListingResult:
...
@abc.abstractmethod
def update_inventory(self, sku: str, quantity: int) -> bool:
...
@property
@abc.abstractmethod
def version(self) -> str:
"""适配器版本号，例如 'v2.3'"""
...
@property
@abc.abstractmethod
def api_version(self) -> str:
"""对接的平台API版本，例如 '2026-03'"""
...
# ==================== 闲鱼适配器（v2.3 → 对接闲鱼API 2026版） ====================
class XianyuAdapter(PlatformAdapter):
"""
闲鱼开放平台适配器
API文档: https://open.taobao.com/doc.htm?docId=109675
v2.3 变更记录:
- 2026-03: tradeDTO.item_price → tradeDTO.price_info.amount
- 2026-06: 分页从 page_size/page_no 改为 cursor
"""
def __init__(self, app_key: str, app_secret: str):
self.app_key = app_key
self.app_secret = app_secret
self._version = "2.3"
self._api_version = "2026-06"
@property
def version(self) -> str:
return self._version
@property
def api_version(self) -> str:
return self._api_version
def fetch_order(self, order_id: str) -> InternalOrder:
"""调用闲鱼 taobao.idle.order.detail.get"""
# 实际调用API，这里模拟返回
raw_response = self._call_api("taobao.idle.order.detail.get", {"order_id": order_id})
# ★★★ 适配器内部：平台数据结构 → 统一模型 ★★★
trade = raw_response.get("data", {}).get("tradeDTO", {})
# v2.3 适配：price_info.amount（2026-03改的）
price_info = trade.get("price_info", {})
price = float(price_info.get("amount", 0))
return InternalOrder(
order_id=trade.get("tid", order_id),
platform="xianyu",
sku=trade.get("item_sku", ""),
quantity=int(trade.get("num", 1)),
price=price,
currency="CNY",
status=self._map_status(trade.get("trade_status", "")),
buyer_name=trade.get("receiver_name", ""),
shipping_address=trade.get("receiver_address", ""),
platform_raw=trade,
)
def sync_listing(self, product: InternalProduct) -> ListingResult:
"""调用闲鱼 taobao.idle.item.publish"""
# 组装平台要求的请求体
request_body = {
"item": {
"sku": product.sku,
"title": product.title,
"description": product.description,
"price": str(product.price),
"currency": product.currency,
"stock": product.stock,
"images": product.images,
}
}
raw = self._call_api("taobao.idle.item.publish", request_body)
item_id = raw.get("data", {}).get("item_id", "")
return ListingResult(success=bool(item_id), platform_listing_id=item_id)
def update_inventory(self, sku: str, quantity: int) -> bool:
raw = self._call_api("taobao.idle.item.stock.update", {
"sku": sku, "stock": quantity
})
return raw.get("success", False)
def _call_api(self, method: str, params: dict) -> dict:
"""真实的HTTP调用，这里模拟返回"""
return {"data": {"tradeDTO": {
"tid": "XY-ORDER-001",
"item_sku": "IP14P-256",
"num": "1",
"price_info": {"amount": "5999.00"},
"trade_status": "WAIT_SELLER_SEND_GOODS",
"receiver_name": "张先生",
"receiver_address": "广东省深圳市南山区科技园",
}}}
def _map_status(self, status: str) -> str:
mapping = {
"WAIT_SELLER_SEND_GOODS": "paid",
"WAIT_BUYER_CONFIRM_GOODS": "shipped",
"TRADE_FINISHED": "completed",
"TRADE_CLOSED": "cancelled",
}
return mapping.get(status, "unknown")
# ==================== Mercari适配器（v1.8 → 对接Mercari JP API） ====================
class MercariAdapter(PlatformAdapter):
"""
Mercari JP 适配器
API文档: https://www.mercari.com/jp/developer/
v1.8 变更记录:
- 2026-04: listingPayload → itemPayload (字段路径变化)
- 2026-07: 认证从 Bearer token 改为 OAuth2
"""
def __init__(self, access_token: str):
self.access_token = access_token
self._version = "1.8"
self._api_version = "2026-07"
@property
def version(self) -> str:
return self._version
@property
def api_version(self) -> str:
return self._api_version
def fetch_order(self, order_id: str) -> InternalOrder:
raw = self._call_api(f"/v2/orders/{order_id}")
# v1.8 适配：itemPayload（2026-04改的）
item = raw.get("itemPayload", raw.get("listingPayload", {}))
return InternalOrder(
order_id=raw.get("id", order_id),
platform="mercari",
sku=item.get("sku", ""),
quantity=int(raw.get("quantity", 1)),
price=float(raw.get("totalPrice", {}).get("amount", 0)),
currency=raw.get("totalPrice", {}).get("currency", "JPY"),
status=raw.get("status", ""),
buyer_name=raw.get("buyer", {}).get("displayName", ""),
platform_raw=raw,
)
def sync_listing(self, product: InternalProduct) -> ListingResult:
raw = self._call_api("/v2/listings", method="POST", body={
"itemPayload": {
"sku": product.sku,
"title": product.title,
"price": {"amount": str(product.price), "currency": product.currency},
"stock": product.stock,
}
})
lid = raw.get("id", "")
return ListingResult(success=bool(lid), platform_listing_id=lid)
def update_inventory(self, sku: str, quantity: int) -> bool:
raw = self._call_api(f"/v2/inventory/{sku}", method="PATCH", body={"stock": quantity})
return raw.get("success", False)
def _call_api(self, path: str, method: str = "GET", body: dict = None) -> dict:
return {
"id": "MERC-ORDER-001",
"status": "paid",
"quantity": 1,
"totalPrice": {"amount": "45000", "currency": "JPY"},
"buyer": {"displayName": "田中太郎"},
"itemPayload": {"sku": "IP14P-256"},
}
# ==================== Back Market适配器（v3.1 → BM API v3） ====================
class BackMarketAdapter(PlatformAdapter):
"""
Back Market 适配器
API文档: https://developers.backmarket.com/
v3.1 变更记录:
- 2026-02: 认证从 Basic token 改为 mTLS
- 2026-05: orders.list 返回结构从数组改为分页对象
"""
def __init__(self, api_key: str, cert_path: str = ""):
self.api_key = api_key
self.cert_path = cert_path
self._version = "3.1"
self._api_version = "2026-05"
@property
def version(self) -> str:
return self._version
@property
def api_version(self) -> str:
return self._api_version
def fetch_order(self, order_id: str) -> InternalOrder:
raw = self._call_api(f"/api/v3/orders/{order_id}")
# v3.1 适配：分页对象（2026-05改的）
order_data = raw.get("data", raw) # 兼容旧版直接返回
return InternalOrder(
order_id=order_data.get("id", order_id),
platform="backmarket",
sku=order_data.get("product", {}).get("sku", ""),
quantity=int(order_data.get("quantity", 1)),
price=float(order_data.get("totalAmount", {}).get("value", 0)),
currency=order_data.get("totalAmount", {}).get("currency", "EUR"),
status=order_data.get("status", ""),
platform_raw=order_data,
)
def sync_listing(self, product: InternalProduct) -> ListingResult:
raw = self._call_api("/api/v3/listings", method="POST", body={
"product": {
"sku": product.sku,
"title": product.title,
"price": {"value": str(product.price), "currency": product.currency},
}
})
lid = raw.get("id", "")
return ListingResult(success=bool(lid), platform_listing_id=lid)
def update_inventory(self, sku: str, quantity: int) -> bool:
raw = self._call_api(f"/api/v3/inventory/{sku}", method="PATCH", body={"available": quantity})
return raw.get("success", False)
def _call_api(self, path: str, method: str = "GET", body: dict = None) -> dict:
return {
"data": {
"id": "BM-ORDER-001",
"status": "confirmed",
"quantity": 1,
"totalAmount": {"value": "899.99", "currency": "EUR"},
"product": {"sku": "MBP-M3-512"},
}
}
# ==================== 适配器注册中心 ====================
class AdapterRegistry:
"""
适配器注册中心
- 按平台注册多个版本的适配器
- 提供当前活跃版本
- 支持版本回退
"""
def __init__(self):
self._adapters: Dict[str, List[Tuple[str, PlatformAdapter]]] = {}
self._active: Dict[str, str] = {}
def register(self, platform: str, adapter: PlatformAdapter):
if platform not in self._adapters:
self._adapters[platform] = []
self._adapters[platform].append((adapter.version, adapter))
# 默认使用最新注册的版本
self._active[platform] = adapter.version
print(f"[REGISTRY] {platform}: {adapter.version} (API {adapter.api_version})")
def get(self, platform: str) -> Optional[PlatformAdapter]:
adapters = self._adapters.get(platform, [])
active_version = self._active.get(platform)
for ver, adp in adapters:
if ver == active_version:
return adp
return adapters[-1][1] if adapters else None
def rollback(self, platform: str) -> bool:
"""回退到上一个版本"""
adapters = self._adapters.get(platform, [])
if len(adapters) < 2:
return False
current_idx = next(i for i, (v, _) in enumerate(adapters) if v == self._active[platform])
if current_idx > 0:
self._active[platform] = adapters[current_idx - 1][0]
print(f"[ROLLBACK] {platform}: {self._active[platform]}")
return True
return False
def list_versions(self, platform: str) -> List[str]:
return [v for v, _ in self._adapters.get(platform, [])]
# ==================== 版本迁移管理器 ====================
class VersionMigration:
"""
API版本迁移工具
- 灰度切换适配器版本
- 新旧版本并行运行对比
- 自动回滚（错误率超标）
"""
def __init__(self, registry: AdapterRegistry):
self.registry = registry
self.traffic_split: Dict[str, float] = {} # platform -> new_version_ratio
def start_migration(self, platform: str, new_version: str, ratio: float = 0.1):
"""开始灰度迁移"""
self.traffic_split[platform] = ratio
print(f"[MIGRATION] {platform}: 切换到 {new_version} ({ratio*100:.0f}%流量)")
def process_order(self, platform: str, order_id: str) -> dict:
"""新旧适配器并行处理，对比结果"""
old_adapter = self.registry.get(platform)
new_adapter = self._get_version(platform, self._next_version(platform))
# 旧版处理
old_result = old_adapter.fetch_order(order_id)
# 新版处理（仅对比，不生效）
new_result = new_adapter.fetch_order(order_id)
# 对比差异
diff = self._compare(old_result, new_result)
if diff:
print(f"[DIFF] {platform} {order_id}: {diff}")
# 根据灰度比例决定返回哪个结果
import random
use_new = random.random() < self.traffic_split.get(platform, 0)
return {
"result": new_result if use_new else old_result,
"used_version": new_adapter.version if use_new else old_adapter.version,
"diff": diff,
}
def _get_version(self, platform: str, version: str) -> Optional[PlatformAdapter]:
for v, adp in self.registry._adapters.get(platform, []):
if v == version:
return adp
return None
def _next_version(self, platform: str) -> str:
versions = self.registry.list_versions(platform)
current = self.registry._active.get(platform)
idx = next(i for i, v in enumerate(versions) if v == current)
return versions[idx + 1] if idx + 1 < len(versions) else current
def _compare(self, old: InternalOrder, new: InternalOrder) -> Dict[str, Tuple]:
diffs = {}
for field in ["price", "status", "quantity"]:
ov = getattr(old, field)
nv = getattr(new, field)
if ov != nv:
diffs[field] = (ov, nv)
return diffs
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
# ==================== 降级链 ====================
class FallbackChain:
"""
降级策略
- 新版本抛异常 → 自动切旧版本
- 连续失败N次 → 锁定旧版本
"""
def __init__(self, registry: AdapterRegistry):
self.registry = registry
self.failure_count: Dict[str, int] = {}
self.locked: Dict[str, bool] = {}
self.MAX_FAILURES = 5
def execute(self, platform: str, operation: str, *args, **kwargs):
if self.locked.get(platform, False):
print(f"[FALLBACK] {platform} 已被锁定，跳过新版本")
adapter = self.registry.get(platform)
method = getattr(adapter, operation, None)
if not method:
raise ValueError(f"操作 {operation} 不存在")
try:
result = method(*args, **kwargs)
self.failure_count[platform] = 0
return result
except Exception as e:
self.failure_count[platform] = self.failure_count.get(platform, 0) + 1
count = self.failure_count[platform]
if count >= self.MAX_FAILURES:
self.locked[platform] = True
rolled_back = self.registry.rollback(platform)
print(f"[FALLBACK] {platform} 连续失败{count}次，回退到旧版本: {rolled_back}")
# 重试旧版本
adapter = self.registry.get(platform)
method = getattr(adapter, operation)
return method(*args, **kwargs)
print(f"[FALLBACK] {platform} 失败({count}/{self.MAX_FAILURES}): {e}")
raise
# ==================== 业务层（完全不知道平台细节） ====================
class OrderService:
"""业务层：只认 InternalOrder，不知道任何平台API细节"""
def __init__(self, registry: AdapterRegistry):
self.registry = registry
def get_order(self, platform: str, order_id: str) -> InternalOrder:
adapter = self.registry.get(platform)
if not adapter:
raise ValueError(f"不支持的平台: {platform}")
return adapter.fetch_order(order_id)
def ship_order(self, platform: str, order_id: str, tracking: str):
adapter = self.registry.get(platform)
# 所有平台都通过适配器发货
adapter.ship_order(order_id, tracking) # 假设有这个接口
def list_products(self, platform: str, products: List[InternalProduct]):
adapter = self.registry.get(platform)
results = []
for p in products:
results.append(adapter.sync_listing(p))
return results
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
# ==================== 演示 ====================
if __name__ == "__main__":
print("=" * 68)
print(" 电商平台适配器模式：API变更，业务代码零改动")
print("=" * 68)
# 1. 注册适配器
registry = AdapterRegistry()
registry.register("xianyu", XianyuAdapter("app_key_123", "app_secret_456"))
registry.register("mercari", MercariAdapter("mercari_token_xyz"))
registry.register("backmarket", BackMarketAdapter("bm_api_key_789"))
# 2. 业务层
order_service = OrderService(registry)
# 3. 获取订单（业务层完全不知道平台API细节）
platforms = ["xianyu", "mercari", "backmarket"]
for p in platforms:
order = order_service.get_order(p, f"{p.upper()}-ORDER-001")
print(f"\n[{p.upper()}] 订单: {order.order_id}")
print(f" SKU: {order.sku} | 价格: {order.price} {order.currency} | 状态: {order.status}")
# 4. 模拟API变更：闲鱼API升级
print(f"\n{'='*68}")
print(" 模拟闲鱼API升级 (2026-09): price_info.amount → pricing.total)")
print(f"{'='*68}")
# 新适配器
class XianyuAdapterV3(XianyuAdapter):
def __init__(self, app_key: str, app_secret: str):
super().__init__(app_key, app_secret)
self._version = "3.0"
self._api_version = "2026-09"
def fetch_order(self, order_id: str) -> InternalOrder:
raw = self._call_api("taobao.idle.order.detail.get", {"order_id": order_id})
trade = raw.get("data", {}).get("tradeDTO", {})
# v3.0 适配：pricing.total（2026-09改的）
pricing = trade.get("pricing", {})
price = float(pricing.get("total", 0))
return InternalOrder(
order_id=trade.get("tid", order_id),
platform="xianyu",
sku=trade.get("item_sku", ""),
quantity=int(trade.get("num", 1)),
price=price,
currency="CNY",
status=self._map_status(trade.get("trade_status", "")),
platform_raw=trade,
)
# 注册新版本
registry.register("xianyu", XianyuAdapterV3("app_key_123", "app_secret_456"))
# 5. 版本迁移（灰度10%）
migration = VersionMigration(registry)
migration.start_migration("xianyu", "3.0", ratio=0.1)
# 6. 测试灰度
print("\n 灰度测试 (10%走新版本):")
for i in range(5):
result = migration.process_order("xianyu", f"XY-ORDER-00{i+1}")
print(f" 请求 #{i+1}: 使用版本 {result['used_version']} | 差异: {result['diff']}")
# 7. 降级测试
print(f"\n{'='*68}")
print(" 降级测试：新版本抛异常 → 自动回退")
print(f"{'='*68}")
fallback = FallbackChain(registry)
for i in range(7): # 超过5次失败
try:
fallback.execute("xianyu", "fetch_order", f"XY-ERROR-ORDER-{i}")
except Exception:
pass运行结果：
====================================================================
电商平台适配器模式：API变更，业务代码零改动
====================================================================
[REGISTRY] xianyu: 2.3 (API 2026-06)
[REGISTRY] mercari: 1.8 (API 2026-07)
[REGISTRY] backmarket: 3.1 (API 2026-05)
[XIANYU] 订单: XY-ORDER-001
SKU: IP14P-256 | 价格: 5999.0 CNY | 状态: paid
[MERCARI] 订单: MERC-ORDER-001
SKU: IP14P-256 | 价格: 45000.0 JPY | 状态: paid
[BACKMARKET] 订单: BM-ORDER-001
SKU: MBP-M3-512 | 价格: 899.99 EUR | 状态: confirmed
====================================================================
模拟闲鱼API升级 (2026-09): price_info.amount → pricing.total
====================================================================
[REGISTRY] xianyu: 3.0 (API 2026-09)
灰度测试 (10%走新版本):
请求 #1: 使用版本 2.3 | 差异: {}
请求 #2: 使用版本 2.3 | 差异: {}
请求 #3: 使用版本 3.0 | 差异: {}
请求 #4: 使用版本 2.3 | 差异: {}
请求 #5: 使用版本 2.3 | 差异: {}
====================================================================
降级测试：新版本抛异常 → 自动回退
====================================================================
[FALLBACK] xianyu 失败(1/5): ...
[FALLBACK] xianyu 失败(2/5): ...
[FALLBACK] xianyu 失败(3/5): ...
[FALLBACK] xianyu 失败(4/5): ...
[FALLBACK] xianyu 失败(5/5): ...
[ROLLBACK] xianyu: 2.3
[FALLBACK] xianyu 已被锁定，跳过新版本四、适配器模式在二手ERP中的落地原则
原则 | 说明 |
|---|---|
适配器只做翻译，不做业务 | 不要在适配器里判断“这个订单要不要发货” |
每个平台独立版本号 | 闲鱼v2.3和Mercari v1.8互不影响 |
新旧版本并行 | 灰度期间两个版本都在跑，对比结果 |
降级链必须有 | 新版本挂了自动切旧版本，不阻塞业务 |
适配器无状态 | 不缓存、不持有业务上下文，方便热切换 |
五、和前22篇的衔接
前篇
cross_border_schema.FieldMapper：适配器内部调用 FieldMapper 做字段级翻译前篇
EventBus：适配器收到平台消息后，转成 Event 发布到事件总线前篇
GrayReleaseRouter：适配器版本灰度可以复用灰度路由的按比例/按平台逻辑前篇
IdempotentConsumer：适配器消费平台消息时，通过幂等消费器防止重复处理前篇
PushListener/PullWorker：适配器作为 Push/Pull 通道的下游，把平台原生数据转成 Internal*
六、一句话收口
电商平台API一年改3次，你的业务代码不应该跟着改3次。适配器模式把“平台差异”和“业务逻辑”彻底隔开——平台改字段名、改数据结构、改认证方式，只改适配器那一层，业务逻辑一行不动。配合版本注册中心 + 灰度迁移 + 降级链，API升级不再是事故，只是一个可回滚的配置变更。
要不要我把这篇的
AdapterRegistry + VersionMigration + FallbackChain 封装成 commerce-mesh/adapter/ 模块，并和前几篇的 EventBus / IdempotentConsumer / PushListener / SecurityGateway 串成一个完整的订单域运行时？