问题描述
我正在研究 Coingecko Api Wrapper。似乎我有一个小问题 - 也许不是问题，但我觉得有更好、更干净的方法来做到这一点。
因此在 Coingecko API 中，如果您发出任何请求调用，您会将值解析为负载（查询字符串）。这是简单的请求字符串：
https://api.coingecko.com/api/v3/simple/price?ids=bitcoin&vs_currencies=eth&include_market_cap=false&include_24hr_vol=false&include_24hr_change=false&include_last_updated_at=false
所以我在 RestClient 类中编写了一个简单的方法，它接受参数和 kwargs（如果存在）并将它们连接到一个字典中：
def _build_payload(self,params,**kwargs):
del params['self']
payload = {**params,**kwargs} if kwargs else params
return payload
并且我在调用给定端点的 get_price 方法中的 GeckoClient 中使用它。所以正如你在下面看到的，我得到了我使用 locals() 的参数的名称。但老实说，我不知道这样做是不是一个坏方法。
def get_price(self,ids: [str,list],vs_currencies: [str,include_market_cap=False,include_24hr_vol=False,include_24hr_change=False,include_last_updated_at=False,**kwargs
):
payload = self._build_payload(locals(),**kwargs)
return self._request('GET','simple/price',payload=payload)
所以我的问题是：有没有更好的方法可以让所有端点通用的函数，它将从每个方法中获取参数和 kwargs 并为我构建一个有效负载（查询字符串）？我认为我对当地人的态度非常糟糕。
你可以在下面看到我的代码。任何建议和意见都会有很大帮助。谢谢！
# gecko.py
class GeckoClient(RestClient):
def __init__(self,config=GeckoConfig):
super(GeckoClient,self).__init__(config)
def get_price(self,**kwargs
):
"""Params:
* ids string (string) [required] - id of coins,comma-separated if querying more than 1 coin *refers to coins/list
* vs_currencies (string) [required] - vs_currency of coins,comma-separated if querying more than 1 vs_currency*refers to simple/supported_vs_currencies
* include_market_cap (string) [optional] - true/false to include market_cap,default: false
* include_24hr_vol (string) [optional] - true/false to include 24hr_vol,default: false
* include_24hr_change (string) [optional] - true/false to include 24hr_change,default: false
* include_last_updated_at (string) [optional] - true/false to include last_updated_at of price,default: false"""
payload = self._build_payload(locals(),**kwargs)
return self._request('GET',payload=payload)
# restclient.py
class RestClient:
def __init__(self,config: Config):
self._url = config.BASE_URL
self._adapter = config.ADAPTER
self._retry_strategy = config.RETRY_STRATEGY
if config.SESSION:
self.session = requests.Session()
else:
self.session = requests
self.session.mount('http://',self._adapter)
def __repr__(self):
return f'<RestClient(url="{self._url}")>'
def _build_endpoint_url(self,url):
return f"{self._url}{url}"
def _build_payload(self,**kwargs} if kwargs else params
return payload
def _request(self,method: str,url: str,payload=dict):
endpoint_url = self._build_endpoint_url(url)
response = self.session.request(method,endpoint_url,params=payload)
try:
response.raise_for_status()
except HTTPError as e:
logger.error(response.json())
raise RestClientError(e)
print(response.json())
return response.json()
解决方法
暂无找到可以解决该程序问题的有效方法，小编努力寻找整理中！
如果你已经找到好的解决方法，欢迎将解决方案带上本链接一起发送给小编。
小编邮箱:dio#foxmail.com (将#修改为@）