当前位置：首页 > news >正文

Python亚马逊SP-API实战指南：5步构建高效电商自动化系统

news 2026/6/6 21:30:30

Python亚马逊SP-API实战指南：5步构建高效电商自动化系统

【免费下载链接】python-amazon-sp-apiPython wrapper to access the amazon selling partner API项目地址: https://gitcode.com/gh_mirrors/py/python-amazon-sp-api

想要通过Python快速接入亚马逊销售伙伴生态系统，实现订单、库存、报告的自动化管理吗？python-amazon-sp-api库提供了完整的解决方案，让开发者能够高效构建专业的电商数据集成平台。本指南将带您从零开始，掌握这个强大工具的核心使用方法。

📦 项目核心价值与应用场景

python-amazon-sp-api是一个专为亚马逊销售伙伴API设计的Python封装库，它简化了复杂的API调用流程，让开发者能够专注于业务逻辑而非底层通信细节。通过这个库，您可以实现：

订单自动化处理- 实时获取、处理和同步订单数据
库存智能管理- 监控库存水平并自动调整补货策略
销售数据分析- 生成多维度的业务报告和销售洞察
产品信息同步- 批量更新产品列表和价格信息
财务数据整合- 对接亚马逊的财务和支付系统

🔧 环境配置与安装部署

系统要求与依赖准备

开始之前，请确保您的开发环境满足以下要求：

Python 3.8或更高版本
有效的亚马逊开发者账号和应用配置
已获取的LWA凭证和刷新令牌
适当的网络访问权限（部分API需要特定地区的访问）

安装方式选择

根据您的项目需求，可以选择不同的安装方案：

# 基础安装 - 包含核心功能 pip install python-amazon-sp-api # AWS密钥管理支持 - 适合企业级部署 pip install "python-amazon-sp-api[aws]" # AWS缓存支持 - 提升性能表现 pip install "python-amazon-sp-api[aws-caching]"

凭证配置策略

库支持三种凭证加载方式，按优先级从高到低排列：

代码参数传递- 直接在客户端初始化时提供
环境变量设置- 通过系统环境变量配置
配置文件读取- 使用YAML格式的配置文件

推荐使用配置文件方式管理凭证，便于不同环境切换：

# credentials.yml 示例配置 version: '1.0' development: refresh_token: '您的刷新令牌' lwa_app_id: '应用标识符' lwa_client_secret: '客户端密钥' aws_access_key: 'AWS访问密钥' aws_secret_key: 'AWS密钥' role_arn: 'IAM角色ARN'

🚀 亚马逊应用创建与授权流程

开发者控制台配置

在开始编码之前，您需要在亚马逊开发者中心完成应用注册。这是连接API的关键步骤：

创建应用时需要填写以下关键信息：

应用名称：2-40个字符的标识符
API类型：选择SP API（销售伙伴API）
IAM ARN：AWS身份和访问管理角色ARN
权限配置：根据业务需求选择相应的API角色
OAuth设置：配置登录和重定向URI

LWA凭证获取与管理

应用创建后，您将获得LWA（Login with Amazon）凭证，这是API身份验证的核心：

重要凭证包括：

Client Identifier：应用的唯一标识符
Client Secret：客户端密钥，需妥善保管
刷新令牌：用于获取访问令牌的长期凭证

应用授权与市场关联

完成凭证配置后，需要为应用授权并关联目标市场：

授权流程会生成新的刷新令牌，该令牌用于后续所有API调用的身份验证。每个市场需要单独授权，确保API调用具有正确的区域权限。

💻 核心API功能实战示例

订单数据获取与处理

订单管理是电商自动化的核心，以下是获取最近订单的完整示例：

from sp_api.api import Orders from datetime import datetime, timedelta, timezone from sp_api.base import SellingApiException def fetch_recent_orders(days_back=7): """获取指定天数内的订单数据""" try: orders_client = Orders() # 计算时间范围 cutoff_date = datetime.now(timezone.utc) - timedelta(days=days_back) # 调用API获取订单 response = orders_client.get_orders( CreatedAfter=cutoff_date.isoformat(), OrderStatuses=['Shipped', 'Unshipped'], MarketplaceIds=['ATVPDKIKX0DER'] # 美国市场 ) # 处理响应数据 if response.payload and 'Orders' in response.payload: orders = response.payload['Orders'] print(f"成功获取 {len(orders)} 个订单") # 提取关键信息 for order in orders: order_id = order.get('AmazonOrderId', 'N/A') purchase_date = order.get('PurchaseDate', 'N/A') order_status = order.get('OrderStatus', 'N/A') amount = order.get('OrderTotal', {}).get('Amount', '0.00') print(f"订单ID: {order_id}") print(f"购买时间: {purchase_date}") print(f"状态: {order_status}") print(f"金额: ${amount}") print("-" * 40) return response.payload except SellingApiException as ex: print(f"API调用失败: {ex}") return None

库存监控与自动化管理

库存管理对于电商业务至关重要，以下代码展示如何获取库存摘要：

from sp_api.api import Inventories from sp_api.base import Marketplaces def get_inventory_summary(marketplace_id='ATVPDKIKX0DER'): """获取指定市场的库存摘要信息""" try: inventory_client = Inventories() # 设置查询参数 params = { 'marketplaceIds': [marketplace_id], 'details': True, 'granularityType': 'Marketplace', 'granularityId': marketplace_id } # 调用库存API response = inventory_client.get_inventory_summaries(**params) if response.payload: summaries = response.payload.get('inventorySummaries', []) print(f"库存摘要报告 - 市场: {marketplace_id}") print("=" * 60) for summary in summaries: asin = summary.get('asin', 'N/A') fnsku = summary.get('fnSku', 'N/A') condition = summary.get('condition', 'N/A') # 库存数量信息 total_quantity = summary.get('totalQuantity', 0) in_stock_quantity = summary.get('inStockQuantity', 0) print(f"ASIN: {asin}") print(f"FNSKU: {fnsku}") print(f"状态: {condition}") print(f"总库存: {total_quantity}") print(f"在售库存: {in_stock_quantity}") print("-" * 40) return response.payload except Exception as e: print(f"库存查询失败: {e}") return None

报告生成与数据处理

亚马逊的报告系统提供了丰富的业务数据，以下是如何创建和获取报告：

from sp_api.api import Reports from sp_api.base.reportTypes import ReportType import time def create_and_download_report(report_type, marketplace_ids=None): """创建报告并等待完成后下载数据""" if marketplace_ids is None: marketplace_ids = ['ATVPDKIKX0DER'] try: reports_client = Reports() # 第一步：创建报告请求 print(f"开始创建 {report_type} 报告...") create_response = reports_client.create_report( reportType=report_type, marketplaceIds=marketplace_ids, dataStartTime='2024-01-01T00:00:00Z', dataEndTime='2024-12-31T23:59:59Z' ) report_id = create_response.payload.get('reportId') print(f"报告创建成功，ID: {report_id}") # 第二步：轮询报告状态 max_attempts = 30 attempt = 0 while attempt < max_attempts: status_response = reports_client.get_report(report_id) processing_status = status_response.payload.get('processingStatus') if processing_status == 'DONE': print("报告处理完成，准备下载...") break elif processing_status == 'CANCELLED': print("报告处理被取消") return None elif processing_status == 'FATAL': print("报告处理失败") return None else: print(f"报告处理中... 状态: {processing_status}") time.sleep(30) # 等待30秒 attempt += 1 if attempt >= max_attempts: print("报告处理超时") return None # 第三步：获取报告文档 document_response = reports_client.get_report_document(report_id) document_url = document_response.payload.get('url') print(f"报告文档URL: {document_url}") # 这里可以添加下载和处理文档的逻辑 # 注意：实际文档下载需要处理加密和压缩 return { 'report_id': report_id, 'document_url': document_url, 'status': 'COMPLETED' } except Exception as e: print(f"报告处理失败: {e}") return None

🔄 异步API客户端使用指南

异步编程优势

python-amazon-sp-api提供了完整的异步支持，能够显著提升I/O密集型操作的性能。异步客户端位于sp_api.asyncio模块中：

import asyncio from datetime import datetime, timedelta, timezone from sp_api.asyncio.api import Orders, Reports, Inventories from sp_api.base.reportTypes import ReportType async def fetch_multiple_data_sources(): """同时获取多种数据源 - 异步版本""" async with Orders() as orders_client, \ Reports() as reports_client, \ Inventories() as inventory_client: # 并行执行多个API调用 orders_task = orders_client.get_orders( LastUpdatedAfter=(datetime.now(timezone.utc) - timedelta(days=1)).isoformat() ) reports_task = reports_client.create_report( reportType=ReportType.GET_MERCHANT_LISTINGS_ALL_DATA ) inventory_task = inventory_client.get_inventory_summaries( marketplaceIds=['ATVPDKIKX0DER'], details=True ) # 等待所有任务完成 orders_result, reports_result, inventory_result = await asyncio.gather( orders_task, reports_task, inventory_task ) # 处理结果 print(f"获取到 {len(orders_result.payload.get('Orders', []))} 个订单") print(f"报告ID: {reports_result.payload.get('reportId', 'N/A')}") print(f"库存摘要数量: {len(inventory_result.payload.get('inventorySummaries', []))}") return { 'orders': orders_result.payload, 'report': reports_result.payload, 'inventory': inventory_result.payload } # 运行异步函数 if __name__ == "__main__": results = asyncio.run(fetch_multiple_data_sources())

异步客户端配置

异步客户端支持连接池和更高效的网络通信，特别适合高并发场景：

from sp_api.asyncio.base import AsyncClient import httpx async def create_custom_async_client(): """创建自定义配置的异步客户端""" # 自定义传输配置 transport = httpx.AsyncHTTPTransport( retries=3, limits=httpx.Limits( max_connections=100, max_keepalive_connections=20, keepalive_expiry=30.0 ) ) # 创建客户端 async with AsyncClient( credentials={ 'refresh_token': 'your_token', 'lwa_app_id': 'your_app_id', 'lwa_client_secret': 'your_secret' }, marketplace=Marketplaces.US, transport=transport, timeout=30.0 ) as client: # 使用客户端调用API response = await client.get('/orders/v0/orders') return response

🛡️ 安全与合规性最佳实践

PII数据处理

处理个人身份信息时，必须使用受限数据令牌（RDT）：

from sp_api.api import Orders def handle_pii_data(restricted_data_token): """安全处理包含PII的数据""" # 使用RDT令牌创建客户端 orders_client = Orders(restricted_data_token=restricted_data_token) # 安全地获取订单数据 orders = orders_client.get_orders( CreatedAfter='2024-01-01T00:00:00Z', MarketplaceIds=['ATVPDKIKX0DER'] ) # 处理订单数据（自动脱敏PII信息） for order in orders.payload.get('Orders', []): # 这里获取的是脱敏后的数据 order_id = order.get('AmazonOrderId') # ... 其他处理逻辑 return orders.payload

错误处理与重试机制

健壮的错误处理是生产环境的关键：

from sp_api.base import SellingApiException, SellingApiBadRequestException from sp_api.base.exceptions import SellingApiServerException import time def robust_api_call(api_function, max_retries=3, **kwargs): """带重试机制的API调用包装器""" for attempt in range(max_retries): try: response = api_function(**kwargs) return response except SellingApiBadRequestException as e: # 客户端错误，通常不需要重试 print(f"请求参数错误: {e}") raise except SellingApiServerException as e: # 服务器错误，可能需要重试 print(f"服务器错误 (尝试 {attempt + 1}/{max_retries}): {e}") if attempt < max_retries - 1: wait_time = 2 ** attempt # 指数退避 print(f"等待 {wait_time} 秒后重试...") time.sleep(wait_time) else: print("达到最大重试次数") raise except SellingApiException as e: # 其他API异常 print(f"API调用异常: {e}") raise except Exception as e: # 未知异常 print(f"未知错误: {e}") raise return None

📊 性能优化与监控策略

连接池与缓存配置

优化HTTP连接管理可以显著提升性能：

from sp_api.base.client import Client from sp_api.base.credential_provider import CredentialProvider import httpx class OptimizedAPIClient: """优化配置的API客户端""" def __init__(self, credentials_config=None): self.credentials = CredentialProvider(credentials_config) # 配置HTTP传输层 self.transport = httpx.HTTPTransport( retries=3, limits=httpx.Limits( max_connections=50, max_keepalive_connections=20, keepalive_expiry=60.0 ) ) # 创建基础客户端 self.base_client = Client( credentials=self.credentials, marketplace='US', transport=self.transport, timeout=30.0, max_retries=2 ) def get_orders_with_cache(self, cache_key, **kwargs): """带缓存的订单获取""" # 检查缓存 cached_data = self._get_from_cache(cache_key) if cached_data: return cached_data # 调用API orders = self.base_client.call_api( '/orders/v0/orders', 'GET', params=kwargs ) # 存储到缓存 self._save_to_cache(cache_key, orders) return orders def _get_from_cache(self, key): """缓存获取逻辑（示例）""" # 实现您的缓存逻辑 pass def _save_to_cache(self, key, data): """缓存存储逻辑（示例）""" # 实现您的缓存逻辑 pass

批量操作与速率限制

合理处理批量操作和API速率限制：

import asyncio from typing import List from sp_api.asyncio.api import Products class BatchProductProcessor: """批量产品处理器""" def __init__(self, max_concurrent=5): self.max_concurrent = max_concurrent self.semaphore = asyncio.Semaphore(max_concurrent) async def process_products_batch(self, asin_list: List[str]): """批量处理产品信息""" async def fetch_product(asin): async with self.semaphore: async with Products() as client: # 添加延迟以避免速率限制 await asyncio.sleep(0.1) product = await client.get_product(asin) return product # 创建任务列表 tasks = [fetch_product(asin) for asin in asin_list] # 批量执行 results = await asyncio.gather(*tasks, return_exceptions=True) # 处理结果 successful = [] failed = [] for asin, result in zip(asin_list, results): if isinstance(result, Exception): print(f"获取产品 {asin} 失败: {result}") failed.append(asin) else: successful.append(result) return { 'successful': successful, 'failed': failed, 'total': len(asin_list) }

🏗️ 项目架构与模块设计

核心模块结构

python-amazon-sp-api采用模块化设计，主要模块位于sp_api/api/目录下：

sp_api/ ├── api/ # 所有API端点实现 │ ├── orders/ # 订单管理模块 │ ├── inventories/ # 库存管理模块 │ ├── reports/ # 报告系统模块 │ ├── products/ # 产品信息模块 │ ├── finances/ # 财务数据模块 │ └── ... # 其他业务模块 ├── asyncio/ # 异步客户端实现 ├── base/ # 基础类和工具 └── util/ # 实用工具函数

自定义端点生成

项目提供了强大的端点生成工具，可以快速创建新的API客户端：

# 生成新的API端点 make_endpoint https://raw.githubusercontent.com/amzn/selling-partner-api-models/main/models/listings-restrictions-api-model/listingsRestrictions_2021-08-01.json

这个命令会根据亚马逊官方的API模型自动生成完整的Python客户端代码，支持所有标准操作。

🚨 常见问题与解决方案

认证与授权问题

问题：收到"Invalid authentication token"错误

解决方案：

检查刷新令牌是否过期（通常有效期为1年）
验证LWA凭证是否正确配置
确认IAM角色具有必要的API权限
检查请求的市场区域与授权是否匹配

# 验证凭证的示例代码 from sp_api.auth import AccessTokenClient def validate_credentials(): try: client = AccessTokenClient( refresh_token='your_token', lwa_app_id='your_app_id', lwa_client_secret='your_secret' ) token = client.get_access_token() print("凭证验证成功") return True except Exception as e: print(f"凭证验证失败: {e}") return False

速率限制处理

问题：API调用频繁返回429错误

解决方案：

实现指数退避重试机制
使用批处理减少API调用次数
缓存频繁访问的数据
监控API使用配额

import time from functools import wraps def rate_limit_handler(max_calls_per_second=2): """API速率限制装饰器""" def decorator(func): last_called = [0.0] @wraps(func) def wrapper(*args, **kwargs): elapsed = time.time() - last_called[0] min_interval = 1.0 / max_calls_per_second if elapsed < min_interval: sleep_time = min_interval - elapsed time.sleep(sleep_time) last_called[0] = time.time() return func(*args, **kwargs) return wrapper return decorator

📈 部署与生产环境建议

配置管理策略

生产环境建议使用环境变量或配置管理服务：

import os from dotenv import load_dotenv # 加载环境变量 load_dotenv() class ProductionConfig: """生产环境配置""" @staticmethod def get_credentials(): return { 'refresh_token': os.getenv('AMAZON_REFRESH_TOKEN'), 'lwa_app_id': os.getenv('AMAZON_LWA_APP_ID'), 'lwa_client_secret': os.getenv('AMAZON_LWA_CLIENT_SECRET'), 'aws_access_key': os.getenv('AWS_ACCESS_KEY_ID'), 'aws_secret_key': os.getenv('AWS_SECRET_ACCESS_KEY'), 'role_arn': os.getenv('AMAZON_IAM_ROLE_ARN') } @staticmethod def get_api_settings(): return { 'timeout': 30.0, 'max_retries': 3, 'pool_connections': 100, 'pool_maxsize': 20 }

监控与日志记录

完善的监控系统对于生产环境至关重要：

import logging from datetime import datetime class APIMonitor: """API调用监控器""" def __init__(self): self.logger = logging.getLogger('amazon_api') self.logger.setLevel(logging.INFO) # 添加文件处理器 handler = logging.FileHandler('api_monitor.log') formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s') handler.setFormatter(formatter) self.logger.addHandler(handler) def log_api_call(self, endpoint, method, status, duration, response_size=None): """记录API调用信息""" log_data = { 'timestamp': datetime.utcnow().isoformat(), 'endpoint': endpoint, 'method': method, 'status': status, 'duration_ms': duration * 1000, 'response_size': response_size } self.logger.info(f"API调用记录: {log_data}") # 可以在这里添加监控指标上报 self._report_metrics(log_data) def _report_metrics(self, data): """上报监控指标""" # 集成到您的监控系统（如Prometheus、Datadog等） pass