电商数据API开发实战经验分享（实操）

​

在电商开发这条路上摸爬滚打了七八年，从创业公司的小项目到百万级用户的平台开发，踩过的API坑能写本血泪史。今天不聊虚的，就分享点实打实的实战经验和代码，都是熬夜调试出来的“救命稻草”。

刚入行时做个简单的商品比价功能，想着调用几个平台API拉数据就行。结果第一个坑就栽在京东API上——人家要求必须用HTTPS，而且签名算藏在犄角旮旯的文档里。当时对着文档折腾了两天，才把签名逻辑啃明白，最后封装出了这个工具函数：

import hashlib
import hmac
import time

def generate_signature(params, secret_key):
    sorted_params = sorted(params.items(), key=lambda x: x[0])
    query_string = "&".join([f"{k}={v}" for k, v in sorted_params])
    timestamp = str(int(time.time()))
    string_to_sign = f"{query_string}×tamp={timestamp}"
    signature = hmac.new(secret_key.encode('utf-8'), string_to_sign.encode('utf-8'), hashlib.sha256).hexdigest()
    return {**params, "timestamp": timestamp, "signature": signature}

# 使用示例
api_params = {
    "method": "jd.item.get",
    "app_key": "your_app_key",
    "item_id": "12345678"
}
signed_params = generate_signature(api_params, "your_secret_key")

解决了签名问题，又碰上数据格式不兼容的大麻烦。淘宝、拼多多API接口返回的JSON结构完全不同，尤其是商品属性字段命名五花八门。那段时间天天对着三个平台的API文档写适配代码，最后干脆抽象出一个通用解析函数：

def parse_product_data(raw_data, platform):
    if platform == "taobao":
        return {
            "title": raw_data.get("item_title", ""),
            "price": float(raw_data.get("item_price", 0)),
            "stock": int(raw_data.get("item_stock", 0)),
            "images": raw_data.get("item_images", [])
        }
    elif platform == "pdd":
        return {
            "title": raw_data.get("goods_name", ""),
            "price": float(raw_data.get("group_price", 0)),
            "stock": int(raw_data.get("goods_quantity", 0)),
            "images": raw_data.get("goods_image_urls", [])
        }
    return {}

更崩溃的是接口突然改版。有次半夜接到运营电话，说商品详情页数据全乱了。紧急排查发现平台偷偷升级了API版本，老接口返回的数据格式完全变了。从那以后，每次调用API都会加数据校验层：

def validate_product_data(data):
    required_fields = ["title", "price", "stock"]
    for field in required_fields:
        if field not in data:
            raise ValueError(f"Missing required field: {field}")
    return True

缓存这块也踩过不少坑。最开始用简单的内存缓存，结果服务器一重启数据全丢。后来改用Redis，结合redis-py库实现了带过期时间的缓存策略：

import redis

redis_client = redis.Redis(host='localhost', port=6379, db=0)

def get_cached_data(key):
    data = redis_client.get(key)
    return data.decode('utf-8') if data else None

def set_cached_data(key, value, ex=300):
    redis_client.setex(key, ex, value)

# 使用示例
product_key = "product:12345"
cached_data = get_cached_data(product_key)
if cached_data:
    product_info = eval(cached_data)
else:
    product_info = fetch_product_from_api()
    set_cached_data(product_key, str(product_info))

现在接新项目，第一件事就是写个测试脚手架。用pytest搭了个简易测试框架，专门用来验证API调用逻辑：

import pytest
import requests

@pytest.fixture
def api_url():
    return "https://api.example.com/product"

def test_api_call(api_url):
    response = requests.get(api_url, params={"product_id": "12345"})
    assert response.status_code == 200
    assert "title" in response.json()

这些代码片段都是从真实项目里抠出来的，虽然不是完整工程，但每个功能都经过生产环境验证。如果你在API开发中也遇到过奇葩问题，欢迎在评论区聊聊，说不定能一起碰撞出解决方案！

​审核编辑 黄宇