如何写接口适配兼容转换脚本

本文目录导读：

1. 核心设计原则
2. 常见场景示例
3. 特殊场景适配
4. 最佳实践
5. 测试示例

我来详细说明如何编写接口适配兼容转换脚本,从前端到后端不同场景的解决方案。

核心设计原则

// 适配转换的基本原则
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()

编写接口适配转换脚本的关键：

1. 明确版本差异：详细对比不同版本的接口差异
2. 设计转换规则：字段映射、默认值、类型转换
3. 实现灵活适配：支持正向/反向转换
4. 添加错误处理：转换失败时的降级策略
5. 记录变更日志：便于问题排查
6. 充分测试：覆盖各种边缘情况

选择合适的设计模式（策略模式、适配器模式、中间件模式），根据实际场景决定采用配置驱动还是代码驱动的方式实现。