Блог

Я работаю над оболочкой Api Coingecko. И кажется, что у меня есть небольшая проблема — может быть, и не проблема, но я чувствую, что есть гораздо лучшие и более чистые способы сделать это.

Поэтому в API Coingecko, если вы делаете какой-либо запрос, вы анализируете значения как полезную нагрузку (строку запроса).Вот простая строка запроса :

https://api.coingecko.com/api/v3/simple/price?ids=bitcoinamp;vs_currencies=ethamp;include_market_cap=falseamp;include_24hr_vol=falseamp;include_24hr_change=falseamp;include_last_updated_at=false

Поэтому я написал простой метод в своем классе RestClient, который принимает параметры и параметры, если они существуют, и объединяет их в один словарь:

     def _build_payload(self, params,  **kwargs):
        del params['self']
        payload = {**params, **kwargs} if kwargs else params
        return payload
 

И я использую его в своем GeckoClient в методах get_price, которые вызывают заданную конечную точку. Итак, как вы видите ниже, я получаю имена своих параметров, которые я использую locals(). Но, честно говоря, я не знаю, не плохой ли это способ сделать это.

         def get_price(self,
                  ids: [str, list], vs_currencies: [str, list],
                  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)
 

Итак, мой вопрос: есть ли лучший способ иметь функцию, общую для всех конечных точек, которая будет принимать параметры и параметры от каждого метода и создавать полезную нагрузку (строку запроса) для меня ? Я думаю, что мой подход к местным жителям очень плох.

Вы можете увидеть мой код ниже. Любые предложения и советы будут вам очень полезны. Спасибо!

 # gecko.py

class GeckoClient(RestClient):

    def __init__(self, config=GeckoConfig):
        super(GeckoClient, self).__init__(config)

    def get_price(self,
                  ids: [str, list], vs_currencies: [str, list],
                  include_market_cap=False, include_24hr_vol=False,
                  include_24hr_change=False, include_last_updated_at=False, **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', 'simple/price', 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, params,  **kwargs):
        del params['self']
        payload = {**params, **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()