add coingecko and kleros api

src/apis/coingeckoapi.py

@@ -0,0 +1,45 @@
+"""
+CoingeckoAPI for fetching the price in usd of a given token.
+"""
+# pylint: disable=logging-fstring-interpolation
+
+from typing import Optional
+import requests
+from src.helper_functions import get_logger
+from src.constants import (
+    header,
+    REQUEST_TIMEOUT,
+)
+
+
+class CoingeckoAPI:
+    """
+    Class for fetching token prices from Coingecko.
+    """
+
+    def __init__(self) -> None:
+        self.logger = get_logger()
+
+    def get_token_price_in_usd(self, address: str) -> Optional[float]:
+        """
+        Returns the Coingecko price in usd of the given token.
+        """
+        coingecko_url = (
+            "https://api.coingecko.com/api/v3/simple/token_price/ethereum?contract_addresses="
+            + address
+            + "&vs_currencies=usd"
+        )
+        try:
+            coingecko_data = requests.get(
+                coingecko_url,
+                headers=header,
+                timeout=REQUEST_TIMEOUT,
+            )
+            coingecko_rsp = coingecko_data.json()
+            coingecko_price_in_usd = coingecko_rsp[address]["usd"]
+        except requests.RequestException as err:
+            self.logger.warning(
+                f"Connection error while fetching Coingecko price for token {address}, error: {err}"
+            )
+            return None
+        return coingecko_price_in_usd

src/apis/klerosapi.py

@@ -0,0 +1,42 @@
+"""
+KlerosAPI for fetching the Kleros token list.
+"""
+# pylint: disable=logging-fstring-interpolation
+
+import requests
+from src.helper_functions import get_logger
+from src.constants import (
+    header,
+    REQUEST_TIMEOUT,
+)
+
+
+class KlerosAPI:
+    """
+    Class for fetching the Kleros token list.
+    """
+
+    def __init__(self) -> None:
+        self.logger = get_logger()
+
+    def get_token_list(self) -> list[str]:
+        """
+        Returns the Kleros token list.
+        """
+        kleros_url = "http://t2crtokens.eth.link"
+
+        try:
+            kleros_data = requests.get(
+                kleros_url,
+                headers=header,
+                timeout=REQUEST_TIMEOUT,
+            )
+            kleros_rsp = kleros_data.json()
+            kleros_list = []
+            for token in kleros_rsp["tokens"]:
+                kleros_list.append(token["address"].lower())
+        except requests.RequestException as err:
+            self.logger.warning(
+                f"Connection error while fetching the Kleros token list, error: {err}"
+            )
+        return kleros_list

src/monitoring_tests/buffers_monitoring_test.py

@@ -5,6 +5,8 @@
 # pylint: disable=logging-fstring-interpolation
 import requests
 from src.monitoring_tests.base_test import BaseTest
+from src.apis.coingeckoapi import CoingeckoAPI
+from src.apis.klerosapi import KlerosAPI
 from src.constants import (
     BUFFER_INTERVAL,
     header,
@@ -22,6 +24,8 @@ class BuffersMonitoringTest(BaseTest):
 
     def __init__(self) -> None:
         super().__init__()
+        self.coingecko_api = CoingeckoAPI()
+        self.kleros_api = KlerosAPI()
         self.counter: int = 0
 
     def compute_buffers_value(self) -> bool:
@@ -30,28 +34,19 @@ def compute_buffers_value(self) -> bool:
         """
         # get all token balances of the smart contract
         try:
-            resp = requests.get(
+            ethplorer_data = requests.get(
                 "https://api.ethplorer.io/\
                     getAddressInfo/\
                     0x9008D19f58AAbD9eD0D60971565AA8510560ab41?\
                     apiKey=freekey",
                 headers=header,
                 timeout=REQUEST_TIMEOUT,
             )
-            rsp = resp.json()
-
-            kleros_resp = requests.get(
-                "http://t2crtokens.eth.link",
-                headers=header,
-                timeout=REQUEST_TIMEOUT,
-            )
-            kleros_rsp = kleros_resp.json()
-            kleros_list = []
-            for t in kleros_rsp["tokens"]:
-                kleros_list.append(t["address"])
+            ethplorer_rsp = ethplorer_data.json()
+            kleros_list = self.kleros_api.get_token_list()
 
             value_in_usd = 0.0
-            for token in rsp["tokens"]:
+            for token in ethplorer_rsp["tokens"]:
                 if token["tokenInfo"]["address"] not in kleros_list:
                     continue
                 balance = token["balance"]
@@ -65,19 +60,11 @@ def compute_buffers_value(self) -> bool:
                     # smart contract we use a second price feed, from coingecko, to correct in case
                     # the initial price is indeed off
                     if token_buffer_value_in_usd > 10000:
-                        coingecko_resp = requests.get(
-                            "https://api.coingecko.com/\
-                                api/v3/simple/token_price/\
-                                ethereum?contract_addresses="
-                            + token["tokenInfo"]["address"]
-                            + "&vs_currencies=usd",
-                            headers=header,
-                            timeout=REQUEST_TIMEOUT,
+                        coingecko_price_in_usd = (
+                            self.coingecko_api.get_token_price_in_usd(
+                                token["tokenInfo"]["address"]
+                            )
                         )
-                        coingecko_rsp = coingecko_resp.json()
-                        coingecko_price_in_usd = coingecko_rsp[
-                            token["tokenInfo"]["address"]
-                        ]["usd"]
                         coingecko_value_in_usd = (
                             balance / 10**decimals
                         ) * coingecko_price_in_usd
@@ -90,9 +77,9 @@ def compute_buffers_value(self) -> bool:
             else:
                 self.logger.info(log_output)
 
-        except requests.RequestException as err:
+        except Exception as err:
             self.logger.warning(
-                f"Connection error while fetching buffer tokens and prices, error: {err}"
+                f"Error while fetching buffer tokens and prices, error: {err}"
             )
             return False
         return True