本文目录导读：
我来详细说明如何编写接口适配兼容转换脚本,从前端到后端不同场景的解决方案。
核心设计原则
// 适配转换的基本原则 1. 单一职责：每个转换函数只做一件事 2. 可组合性：基础转换可以组合成复杂转换 3. 错误隔离：转换失败不影响主流程 4. 可追溯：记录转换前后的数据变化 5. 向后兼容：新版本保持对旧数据的支持
常见场景示例
前端接口适配转换
// 接口响应数据转换器类
class ApiResponseTransformer {
constructor() {
this.transformers = new Map();
this.defaultOptions = {
strict: false,
log: false
};
}
// 注册转换器
register(apiVersion, transformer) {
this.transformers.set(apiVersion, transformer);
}
// 执行转换
async transform(response, fromVersion, toVersion) {
const transformer = this.transformers.get(`${fromVersion}->${toVersion}`);
if (!transformer && this.options.strict) {
throw new Error(`No transformer found: ${fromVersion}->${toVersion}`);
}
try {
const result = transformer ? await transformer(response) : response;
if (this.options.log) {
console.log(`[Transform] v${fromVersion}->v${toVersion}:`, {
before: response,
after: result
});
}
return result;
} catch (error) {
console.error('Transform failed:', error);
return response; // 降级返回原始数据
}
}
setOptions(options) {
this.options = { ...this.defaultOptions, ...options };
}
}
// 实际使用示例
const transformer = new ApiResponseTransformer();
transformer.setOptions({ log: true });
// 注册v1到v2的转换器
transformer.register('v1->v2', async (response) => {
return {
id: response.userId,
name: response.userName,
email: response.email,
profile: {
avatar: response.avatarUrl,
bio: response.bio || '',
created_at: new Date(response.registerTime).toISOString()
},
// 填充新字段默认值
status: 'active',
permissions: ['basic']
};
});
// 使用
const v1Data = {
userId: 123,
userName: '张三',
email: 'zhang@example.com',
avatarUrl: 'http://example.com/avatar.jpg',
registerTime: '2023-01-01T00:00:00Z'
};
const v2Data = await transformer.transform(v1Data, 'v1', 'v2');
后端API版本兼容
from typing import Any, Dict, Optional, Callable
from datetime import datetime
import json
class ApiVersionAdapter:
"""API版本适配器"""
def __init__(self):
self._adapters: Dict[str, Callable] = {}
self._version_header = 'X-API-Version'
def register(self, version: str, adapter: Callable):
"""注册版本适配器"""
self._adapters[version] = adapter
def adapt(self, version: str, data: Any) -> Any:
"""根据版本适配数据"""
adapter = self._adapters.get(version)
if adapter:
return adapter(data)
return data
def create_router_middleware(self):
"""创建Flask/FastAPI中间件"""
async def middleware(request, call_next):
# 从请求头获取版本号
version = request.headers.get(
self._version_header,
'latest'
)
# 处理请求
response = await call_next(request)
# 适配响应数据
if hasattr(response, 'body'):
try:
data = json.loads(response.body)
adapted_data = self.adapt(version, data)
response.body = json.dumps(adapted_data).encode()
except:
pass
return response
return middleware
# 实际使用
adapter = ApiVersionAdapter()
# 注册v1适配器（向前兼容）
adapter.register('v1', lambda data: {
'error_code': data.get('code', 0),
'error_msg': data.get('message', ''),
'data': data.get('result', {}),
'timestamp': datetime.now().isoformat()
})
# 注册v2适配器
adapter.register('v2', lambda data: {
'status': 'success' if data.get('code') == 0 else 'error',
'data': data.get('result', {}),
'meta': {
'version': '2.0',
'timestamp': datetime.now().isoformat()
}
})
数据模型转换
// TypeScript 数据转换器
interface DataTransformer<T, U> {
transform(data: T): U;
reverse?(data: U): T;
}
class ModelTransformer {
private transformers: Map<string, DataTransformer<any, any>>;
constructor() {
this.transformers = new Map();
}
// 注册转换规则
register<T, U>(
name: string,
forward: (data: T) => U,
reverse?: (data: U) => T
) {
this.transformers.set(name, {
transform: forward,
reverse
});
}
// 执行转换
transform<T, U>(name: string, data: T): U {
const transformer = this.transformers.get(name);
if (!transformer) {
throw new Error(`Transformer '${name}' not found`);
}
return transformer.transform(data);
}
// 反向转换
reverse<T, U>(name: string, data: U): T {
const transformer = this.transformers.get(name);
if (!transformer?.reverse) {
throw new Error(`Reverse transformer '${name}' not found`);
}
return transformer.reverse(data);
}
}
// 使用示例
interface UserV1 {
id: number;
username: string;
email: string;
phone: string;
}
interface UserV2 {
userId: string;
name: string;
contact: {
email: string;
phone?: string;
};
createdAt: string;
}
const transformer = new ModelTransformer();
transformer.register(
'user_v1_to_v2',
// 正向转换
(v1: UserV1): UserV2 => ({
userId: `user_${v1.id}`,
name: v1.username,
contact: {
email: v1.email,
phone: v1.phone
},
createdAt: new Date().toISOString()
}),
// 反向转换
(v2: UserV2): UserV1 => {
const id = parseInt(v2.userId.replace('user_', ''));
return {
id: id || 0,
username: v2.name,
email: v2.contact.email,
phone: v2.contact.phone || ''
};
}
);
特殊场景适配
JSON Schema 格式转换
def json_schema_adapter(old_schema: dict) -> dict:
"""JSON Schema draft 4 转 draft 7"""
schema_map = {
'type': 'https://json-schema.org/draft-07/schema#',
'title': old_schema.get('title', ''),
'description': old_schema.get('description', ''),
'properties': {}
}
# 转换字段
for field_name, field_schema in old_schema.get('properties', {}).items():
schema_map['properties'][field_name] = {
'type': field_schema.get('type', 'string'),
'description': field_schema.get('description', ''),
'examples': [field_schema.get('default', '')] if 'default' in field_schema else []
}
# 添加验证规则
if 'minLength' in field_schema:
schema_map['properties'][field_name]['minLength'] = field_schema['minLength']
if 'maxLength' in field_schema:
schema_map['properties'][field_name]['maxLength'] = field_schema['maxLength']
# 必填字段
schema_map['required'] = old_schema.get('required', [])
return schema_map
日期格式统一
from datetime import datetime
import re
class DateFormatAdapter:
"""日期格式统一适配器"""
DATE_PATTERNS = {
'yyyy-mm-dd': r'\d{4}-\d{2}-\d{2}',
'dd/mm/yyyy': r'\d{2}/\d{2}/\d{4}',
'timestamp': r'\d{10,13}',
'iso8601': r'\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}'
}
@classmethod
def detect_format(cls, date_string: str) -> str:
"""检测日期格式"""
for format_name, pattern in cls.DATE_PATTERNS.items():
if re.match(pattern, date_string):
return format_name
return 'unknown'
@classmethod
def convert(cls, date_string: str, target_format: str = 'iso8601') -> str:
"""转换日期格式"""
source_format = cls.detect_format(date_string)
# 转换为datetime对象
if source_format == 'yyyy-mm-dd':
dt = datetime.strptime(date_string, '%Y-%m-%d')
elif source_format == 'dd/mm/yyyy':
dt = datetime.strptime(date_string, '%d/%m/%Y')
elif source_format == 'timestamp':
ts = int(date_string)
if len(date_string) == 13: # 毫秒
ts //= 1000
dt = datetime.fromtimestamp(ts)
elif source_format == 'iso8601':
dt = datetime.fromisoformat(date_string)
else:
raise ValueError(f'Unknown date format: {date_string}')
# 输出目标格式
if target_format == 'iso8601':
return dt.isoformat()
elif target_format == 'yyyy-mm-dd':
return dt.strftime('%Y-%m-%d')
elif target_format == 'timestamp':
return str(int(dt.timestamp()))
else:
raise ValueError(f'Unknown target format: {target_format}')
最佳实践
配置驱动的转换
# conversion-config.yaml
versions:
v1-to-v2:
fields:
userId:
target: id
transform: "parseInt"
userName:
target: name
avatarUrl:
target: "profile.avatar"
defaults:
status: "active"
createdAt: "Timestamp.now()"
v2-to-v3:
fields:
id:
target: "uuid"
transform: "generateUUID"
name:
target: "displayName"
remove_fields:
- password
- token
监控与日志
import logging
from functools import wraps
logger = logging.getLogger(__name__)
def monitored_transformation(func):
"""监控转换函数的装饰器"""
@wraps(func)
async def wrapper(*args, **kwargs):
start_time = datetime.now()
try:
result = await func(*args, **kwargs)
duration = (datetime.now() - start_time).total_seconds()
logger.info(f"Transformation {func.__name__}:", {
'duration': f'{duration:.3f}s',
'success': True,
'input_size': len(str(args[0])),
'output_size': len(str(result))
})
return result
except Exception as e:
logger.error(f"Transformation failed {func.__name__}: {str(e)}")
raise
return wrapper
测试示例
import unittest
class TestApiAdapter(unittest.TestCase):
def setUp(self):
self.adapter = ApiVersionAdapter()
self.adapter.register('v1', self.v1_adapter)
self.adapter.register('v2', self.v2_adapter)
def v1_adapter(self, data):
return {'user_id': data['id'], 'name': data['username']}
def v2_adapter(self, data):
return {'userId': str(data['id']), 'displayName': data['username']}
def test_v1_to_v2_conversion(self):
input_data = {'id': 1, 'username': '张三'}
result = self.adapter.adapt('v1', input_data)
self.assertEqual(result['user_id'], 1)
self.assertEqual(result['name'], '张三')
def test_missing_adapter(self):
with self.assertRaises(KeyError):
self.adapter.adapt('v3', {})
if __name__ == '__main__':
unittest.main()
编写接口适配转换脚本的关键：
- 明确版本差异：详细对比不同版本的接口差异
- 设计转换规则：字段映射、默认值、类型转换
- 实现灵活适配：支持正向/反向转换
- 添加错误处理：转换失败时的降级策略
- 记录变更日志：便于问题排查
- 充分测试：覆盖各种边缘情况
选择合适的设计模式（策略模式、适配器模式、中间件模式），根据实际场景决定采用配置驱动还是代码驱动的方式实现。
