首页 > 应用开发 > IT综合 > 电商API数据之《程序员受难记》(1)
import hashlibimport hmacimport jsonimport timedef pdd_sign(params, client_secret):
params = {**params, "client_secret": client_secret}
sorted_params = sorted(params.items(), key=lambda x: x[0])
query_str = "".join([f"{k}{v}" for k, v in sorted_params])
sign = hashlib.md5(query_str.encode()).hexdigest().upper()
return {**params, "sign": sign, "timestamp": int(time.time())}# 调用示例api_params = {
"type": "pdd.goods.detail.get",
"client_id": "your_client_id",
"goods_id": 12345678}signed_params = pdd_sign(api_params, "your_secret_key")def safe_parse_price(price_str):
try:
return float(price_str) if price_str else 0
except ValueError:
return 0parsed_data = {
**raw_data,
"price": safe_parse_price(raw_data.get("price", "0"))}import timeclass LeakyBucket: def __init__(self, capacity, rate): self.capacity = capacity self.rate = rate self.tokens = capacity self.last_update = time.time() def consume(self, tokens=1): now = time.time() # 补充令牌 self.tokens = min(self.capacity, self.tokens + (now - self.last_update) * self.rate) self.last_update = now if self.tokens >= tokens: self.tokens -= tokens return True return False# 使用示例bucket = LeakyBucket(capacity=100, rate=20) # 容量100,每秒补充20个令牌if bucket.consume(): # 调用API response = requests.get(api_url)else: time.sleep(0.1) # 等待令牌补充
import redisfrom functools import lru_cache
redis_client = redis.Redis(host='localhost', port=6379, db=0)@lru_cache(maxsize=128) # 本地LRU缓存def get_from_local_cache(key):
# 实现逻辑...
passdef get_from_redis_cache(key):
data = redis_client.get(key)
return data.decode('utf-8') if data else Nonedef set_redis_cache(key, value, ex=300):
redis_client.setex(key, ex, value)def get_product_data(key):
local_data = get_from_local_cache(key)
if local_data:
return local_data
redis_data = get_from_redis_cache(key)
if redis_data:
local_data = json.loads(redis_data)
get_from_local_cache.cache_clear() # 更新本地缓存
return local_data
# 调用API获取数据
api_data = fetch_api_data()
set_redis_cache(key, json.dumps(api_data))
get_from_local_cache(key) # 存入本地缓存
return api_dataimport pytestimport requestsdef test_api_response():
response = requests.get(api_url)
assert response.status_code == 200, f"Status code error: {response.status_code}"
try:
data = response.json()
assert "error_code" not in data, f"API error: {data.get('error_msg')}"
except ValueError:
pytest.fail("Response is not valid JSON")
来自 “ ITPUB博客 ” ,链接:https://blog.itpub.net/70026917/viewspace-3089826/ ,如需转载,请注明出处,否则将追究法律责任。