Python定价API封装调用test3接口无JSON返回问题求助

我编写了一个定价API的Python封装工具，在代码末尾对4个不同的API端点发起方法调用测试。运行后4个接口中有3个可以正常返回JSON响应，仅第三个名为test3的接口异常。我怀疑是f-string中URL的格式编写有误，尝试调整格式后仍然报错，相关代码如下：

import http.client
import urllib.request, urllib.parse, urllib.error
import json
import base64


def loadJson(response):
   body = response.read()
   if body == "" or body is None:
       print(("Empty response found with status " + str(response.status)))
       return {}
   else:
       return json.loads(body)

class data_pricing_client:

   def __init__(self, host, clientId, clientSecret):
       conn = http.client.HTTPSConnection("sso.lukka.tech")
       path = "/oauth2/aus1imo2fqcx5Ik4Q0h8/v1/token"
       encodedData = base64.b64encode(bytes(f"{clientId}:{clientSecret}", "ISO-8859-1")).decode("ascii")
       authHeader = "Basic " + encodedData
       headers = {
           "Authorization": authHeader,
           "Accept": "application/json",
           "Content-Type": "application/x-www-form-urlencoded"
       }
       params = urllib.parse.urlencode({
           "grant_type": "client_credentials",
           "scope": "pricing"
       })

       conn.request("POST", path, params, headers)
       response = conn.getresponse()

       if response.status != 200:
           raise ApiErrorException(response.status, "Failed to get access token")

       self.host = host
       self.accessToken = loadJson(response)["access_token"]

   def default_headers(self):
       return {
           "Content-Type": "application/json",
           "Accept": "application/json",
           "Authorization": "Bearer " + self.accessToken
       }
  
   def send_message(self,path):
       conn = http.client.HTTPSConnection(self.host)
       headers = self.default_headers()
       conn.request("GET", path, None, headers)
       response_json = loadJson(conn.getresponse())
       conn.close()
       return response_json

   def get_available_sources(self):
       path = "/v1/pricing/sources"
       return self.send_message(path)

   def get_source_details(self, sourceId):
       path = f"/v1/pricing/sources/{sourceId}/prices"
       return self.send_message(path)

   def get_latest_prices(self, sourceId, pairCodes=None, asOf=None, variances=False):
       query_params = {
           "pairCodes":pairCodes,
           "asOf":asOf
       }
       path = f"/v1/pricing/sources/{sourceId}/prices/{'/variances' if variances else ''}?{urllib.parse.urlencode({k: v for k, v in query_params.items() if v is not None})}"
       print(path)
       return self.send_message(path)
  
   def get_historical_prices(self, sourceId, pairCode, from_ts=None, to_ts=None, fill=None, limit=None, variances=False):
       query_params = {
           "from":from_ts,
           "to":to_ts,
           "fill": fill,
           "limit": limit
       }
       path = f"/v1/pricing/sources/{sourceId}/prices/pairs/{pairCode}{'/variances' if variances else ''}?{urllib.parse.urlencode({k: v for k, v in query_params.items() if v is not None})}"
       print(path)
       return self.send_message(path)

       

class ApiErrorException(Exception):
   def __init__(self, status, msg):
       self.msg = "Error " + str(status) + ": " + msg
   def __str__(self):
       return self.msg

if __name__ == '__main__':
   from pricing_api_creds import lukka_pricing
   c = data_pricing_client(**lukka_pricing)

   test1 = c.get_available_sources()
   test1 = c.send_message("/v1/pricing/sources")
   print(test1)

   test2 = c.get_source_details(3000)
   test2 = c.send_message("/v1/pricing/sources/")
   print(test2)

   test3 = c.get_latest_prices(3000)
   test3 = c.send_message("v1/pricing/sources/1000/prices")
   print(test3)

   test4 = c.get_historical_prices(1000, "XBT-USD",)
   test4=c.send_message("/v1/pricing/sources/1000/prices/pairs/XBT-USD")
   print(test4)

核心错误点 #

• test3的手动测试代码中，调用send_message传入的路径缺少开头的/：test3 = c.send_message("v1/pricing/sources/1000/prices")，缺少开头斜杠的路径会被HTTP客户端识别为相对路径，最终拼接出的请求地址不符合服务端的路由规则，这是test3报错的直接原因。
• get_latest_prices和get_historical_prices两个方法的路径拼接逻辑存在瑕疵：当没有有效查询参数时，路径末尾会多余一个?字符，部分服务端接口会无法识别这种格式的路径，也可能引发异常。

修复方法 #

1. 修改test3的调用路径，补全开头的斜杠：

test3 = c.send_message("/v1/pricing/sources/1000/prices")

2. 优化两个接口的路径拼接逻辑，避免多余的?，以get_latest_prices为例，修改后代码如下：

def get_latest_prices(self, sourceId, pairCodes=None, asOf=None, variances=False):
    query_params = {
        "pairCodes":pairCodes,
        "asOf":asOf
    }
    base_path = f"/v1/pricing/sources/{sourceId}/prices/{'/variances' if variances else ''}"
    # 过滤空参数
    valid_params = {k: v for k, v in query_params.items() if v is not None}
    # 仅当有有效参数时拼接查询串
    if valid_params:
        path = f"{base_path}?{urllib.parse.urlencode(valid_params)}"
    else:
        path = base_path
    print(path)
    return self.send_message(path)

get_historical_prices方法可以使用相同逻辑修改，避免末尾多余的?。

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