使用Session Mount结合AWS API Gateway调用接口遇400错误及优化问询

一、400错误的核心原因与修复 #

你遇到的400错误大概率是ApiGateway适配器初始化参数错误导致的：

• 你把完整的接口路径https://my.com/plp_search_v2传给了ApiGateway构造函数，但这类适配器只需要接收域名（https://my.com）即可。挂载后，所有该域名下的请求都会通过适配器处理，接口路径应该由后续的请求URL指定。
• 同时挂载http://和https://可能会干扰HTTPS请求的处理（如果你的目标接口都是HTTPS），没必要保留HTTP的挂载。

修复后的基础请求代码 #

from requests import Session
# 假设你的ApiGateway类来自合适的库，比如aws-requests-auth或自定义实现
gateway = ApiGateway(
    "https://my.com",  # 只传域名，去掉接口路径
    access_key_id=aws_access_key_id,
    access_key_secret=aws_secret_access_id,
    regions=EXTRA_REGIONS
)

url = "https://my.com/plp_search_v2"
headers = {
    'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/92.0.4515.159 Safari/537.36'
}
params = {
    "key": "9f36aeafbe60771e321a7cc95a78140772ab3e96",
    "category": "ziogr",
    "channel": "WEB",
    "count": "24",
}

session = Session()
session.mount("https://", gateway)  # 只挂载HTTPS，去掉HTTP的挂载
response = session.get(url, params=params, headers=headers)
print(response.status_code)

如果修复后仍有400错误，需要检查：

• ApiGateway适配器的签名逻辑是否正确处理了请求路径、参数和方法（GET）
• 确保请求参数没有被适配器意外修改，比如参数编码问题

二、多线程批量调用优化 #

requests的Session不是线程安全的，不能多个线程共享同一个Session实例，否则会出现请求混乱或报错。推荐用concurrent.futures.ThreadPoolExecutor，每个线程创建独立的Session：

多线程实现代码 #

import concurrent.futures
from requests import Session

def make_request(api_gateway_config, url, headers, params):
    # 每个线程创建自己的Session和适配器
    session = Session()
    gateway = ApiGateway(
        api_gateway_config["domain"],
        access_key_id=api_gateway_config["access_key"],
        access_key_secret=api_gateway_config["access_secret"],
        regions=api_gateway_config["regions"]
    )
    session.mount("https://", gateway)
    try:
        response = session.get(url, params=params, headers=headers)
        response.raise_for_status()  # 主动抛出HTTP错误
        return {"url": url, "status": response.status_code, "data": response.json()}
    except Exception as e:
        return {"url": url, "error": str(e)}

# 批量调用配置
api_gateway_config = {
    "domain": "https://my.com",
    "access_key": aws_access_key_id,
    "access_secret": aws_secret_access_id,
    "regions": EXTRA_REGIONS
}
url = "https://my.com/plp_search_v2"
headers = {
    'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/92.0.4515.159 Safari/537.36'
}
# 构造多个不同的params，实现批量调用不同参数的接口
params_list = [
    {"key": "9f36aeafbe60771e321a7cc95a78140772ab3e96", "category": "ziogr", "channel": "WEB", "count": "24"},
    {"key": "9f36aeafbe60771e321a7cc95a78140772ab3e96", "category": "other", "channel": "WEB", "count": "24"},
    # 更多参数组合...
]

# 线程池调用，线程数根据API并发限制调整，比如10-20
with concurrent.futures.ThreadPoolExecutor(max_workers=10) as executor:
    # 提交所有任务
    futures = [executor.submit(make_request, api_gateway_config, url, headers, params) for params in params_list]
    # 收集结果
    for future in concurrent.futures.as_completed(futures):
        result = future.result()
        if "error" in result:
            print(f"请求失败: {result['url']} - {result['error']}")
        else:
            print(f"请求成功: {result['url']} - {result['status']}")

额外优化建议 #

• 如果API有并发限制，不要把max_workers设得太高，避免触发限流
• 给每个请求添加超时时间，比如session.get(..., timeout=(3, 10))，防止线程挂起
• 对于大量请求，可结合HTTPAdapter设置连接池大小，提升连接复用效率：

  from requests.adapters import HTTPAdapter
  adapter = HTTPAdapter(pool_connections=10, pool_maxsize=10)
  session.mount("https://", adapter)
  

内容的提问来源于stack exchange，提问作者bittrago