基于Okta认证的定价API Python封装缺少什么才能获取URL的JSON响应

现有代码核心问题 #

• 实例方法调用错误：get_available_sources等方法调用_send_request时未加self.前缀，会触发未定义变量报错
• 响应未解析为JSON：_send_request当前直接返回原生HTTPResponse对象，没有调用已封装的loadJson方法解析为JSON格式
• URL路径格式错误：get_latest_prices、historical_prices方法用空格拼接参数，不符合HTTP URL规范，需要改为标准查询参数格式
• Okta认证请求未释放连接：初始化阶段调用sso.lukka.tech的请求没有关闭连接，存在资源泄漏风险

具体修改步骤 #

1. 调试Okta认证响应 #

如果需要打印Okta身份认证接口的返回结果，在__init__方法中获取到response后添加打印语句即可：

# 原有代码
conn.request("POST", path, params, headers)
response = conn.getresponse()
# 新增打印语句，直接输出Okta返回的完整JSON内容
print("Okta认证响应：", loadJson(response))

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

2. 修复_send_request方法，自动解析JSON #

def _send_request(self,path):
    with http.client.HTTPSConnection(self.host) as conn:
        headers = self.default_headers()
        conn.request("GET", path, None, headers)
        response = conn.getresponse()
        # 直接返回解析后的JSON结果，不需要再单独处理响应
        return loadJson(response)

3. 修正实例方法调用 #

所有调用_send_request的方法都需要添加self.前缀，例如：

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

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

4. 修复URL路径拼接 #

以get_latest_prices为例，改为标准查询参数拼接方式：

def get_latest_prices(self,asOf, sourceId, pairCodes):
    query_params = urllib.parse.urlencode({
        "asOf": asOf,
        "pairCodes": pairCodes
    })
    path = f"/v1/pricing/sources/{sourceId}?{query_params}"
    return self._send_request(path)

5. 调用接口打印返回结果 #

在main块中调用接口即可直接打印JSON格式的返回内容：

if __name__ == '__main__':
    from pricing_api_creds import lukka_pricing
    c = DataPricingClient(**lukka_pricing)
    # 示例：打印可用数据源接口的返回结果
    sources = c.get_available_sources()
    print("可用数据源：", sources)

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