Add new coingecko data endpoints

run.py

@@ -2,6 +2,13 @@ import json
 import pprint
 import asyncio
 
-import src.knowledge_bases.pipeline as Pipeline
+# import src.knowledge_bases.pipeline as Pipeline
 
-asyncio.run(Pipeline.main())
+# asyncio.run(Pipeline.main())
+
+from src.data_sources.coin_gecko import CoinGecko
+from src.data_sources.cryptocompare import CryptoCompare
+
+cgc = CoinGecko()
+
+pprint.pprint(cgc.get_trending_coin_list())

src/data_sources/coin_gecko.py

@@ -183,3 +183,84 @@ class CoinGecko:
 
         return result
     
+    @redis_cache(ttl=ONE_MONTH_IN_SECONDS)
+    @logger.instrument()
+    def get_coin_categories_list(self, cache_ttl: int = None) -> dict:
+        """
+        Retrieves a list of all available coin categories.
+
+        Args:
+            cache_ttl (int, optional): The time-to-live (in seconds) for the cached data. Defaults to None.
+
+        Returns:
+            dict: A dictionary containing a list of all available coin categories.
+
+        Example:
+            To get a list of all available coin categories, you can call:
+
+            ```python
+            category_list = coin_gecko.get_coin_categories_list()
+            ```
+
+        Note:
+            The `get_coin_categories_list` method uses the `redis_cache` decorator to cache the result for a specified time (`ONE_MONTH_IN_SECONDS`). This means that if the same request is made within the cache time, the cached result will be returned instead of making a new API call.
+        """
+        result = self.cgc.categories.get()
+        logger.debug(f"Result: {result}")
+
+        return { 'exchanges': result }
+    
+    @redis_cache(ttl=ONE_HOUR_IN_SECONDS)
+    @logger.instrument()
+    def get_asset_platforms_list(self, cache_ttl: int = None) -> dict:
+        """
+        Retrieves a list of all available asset platforms.
+
+        Args:
+            cache_ttl (int, optional): The time-to-live (in seconds) for the cached data. Defaults to None.
+
+        Returns:
+            dict: A dictionary containing a list of all available asset platforms.
+
+        Example:
+            To get a list of all available asset platforms, you can call:
+
+            ```python
+            asset_platforms_list = coin_gecko.get_asset_platforms_list()
+            ```
+
+        Note:
+            The `get_asset_platforms_list` method uses the `redis_cache` decorator to cache the result for a specified time (`ONE_HOUR_IN_SECONDS`). This means that if the same request is made within the cache time, the cached result will be returned instead of making a new API call.
+        """
+        result = self.cgc.asset_platform.get()
+        logger.debug(f"Result: {result}")
+
+        return { 'exchanges': result }
+    
+    @redis_cache(ttl=ONE_HOUR_IN_SECONDS)
+    @logger.instrument()
+    def get_trending_coin_list(self, cache_ttl: int = None) -> dict:
+        """
+        Retrieves a list of query trending 15 coins, 7 nfts and 5 categories on CoinGecko in the last 24 hours.
+
+        Args:
+            cache_ttl (int, optional): The time-to-live (in seconds) for the cached data. Defaults to None.
+
+        Returns:
+            dict: A dictionary containing a list of trending coins.
+
+        Example:
+            To get a list of trending coins, you can call:
+
+            ```python
+            trending_coins = coin_gecko.get_trending_coin_list()
+            ```
+
+        Note:
+            The `get_trending_coin_list` method uses the `redis_cache` decorator to cache the result for a specified time (`ONE_HOUR_IN_SECONDS`). This means that if the same request is made within the cache time, the cached result will be returned instead of making a new API call.
+        """
+        result = self.cgc.search.get_trending()
+        logger.debug(f"Result: {result}")
+
+        return { 'exchanges': result }
+    

src/knowledge_bases/pipeline.py

@@ -29,10 +29,13 @@ async def fetch_and_load_crypto_compare_json_data():
 
 async def fetch_and_load_coin_gecko_json_data():
     get_all_coins = make_async(coin_gecko.get_exchanges_list)
+    get_asset_platforms_list = make_async(coin_gecko.get_asset_platforms_list)
 
     all_exchange_data = await get_all_coins()
+    asset_platforms_list = await get_asset_platforms_list()
 
     await store_json_data_in_s3(all_exchange_data, "coin_gecko_all_exchange_data")
+    await store_json_data_in_s3(asset_platforms_list, "coin_gecko_asset_platforms_list")
 
 async def load_json_data_into_knowledge_base():
     load_knowledge_base = make_async(json_knowledge_base.load_knowledge_base)