src/data_sources/coin_gecko.py

import os
from dotenv import load_dotenv
from src.databases.redis import REDIS_CACHED
from src.libs.constants import ONE_HOUR_IN_SECONDS, ONE_MINUTE_IN_SECONDS, ONE_MONTH_IN_SECONDS
from src.libs.logger import logger
from coingecko import CoinGeckoProClient, CoinGeckoDemoClient

load_dotenv()

redis_cache = REDIS_CACHED

class CoinGecko:
    def __init__(self, pro_api: bool = False) -> None:
        """
        Initializes the CoinGecko object.

        Args:
            pro_api (bool, optional): A boolean value indicating whether to use the Pro API. Defaults to False.

        Returns:
            None

        Raises:
            ValueError: If the `pro_api` argument is not a boolean value.

        Note:
            This method initializes the CoinGecko object by creating an instance of either CoinGeckoProClient or CoinGeckoDemoClient based on the `pro_api` argument. If `pro_api` is `True`, it uses the Pro API with the provided API key; otherwise, it uses the Demo API.
        """
        if not isinstance(pro_api, bool):
            raise ValueError("The `pro_api` argument must be a boolean value.")

        if pro_api:
            self.cgc = CoinGeckoProClient(api_key=os.getenv('COINGECKO_PRO_API_KEY'))
        else:
            self.cgc = CoinGeckoDemoClient(api_key=os.getenv('COINGECKO_DEMO_API_KEY'))
    
    @redis_cache(ttl=ONE_MINUTE_IN_SECONDS)
    @logger.instrument()
    def get_coin_price(self, ids: list[str], vs_currencies: list[str], cache_ttl: int = None) -> dict:
        """
        Retrieves the current price of a list of coins in a specified currency.

        Args:
            ids (list[str]): A list of unique identifiers for the coins to retrieve prices for.
            vs_currencies (list[str]): A list of currency codes to convert the coin prices to.
            cache_ttl (int, optional): The time-to-live (in seconds) for the cached data. Defaults to None.

        Returns:
            dict: A dictionary containing the current prices of the specified coins in the specified currencies.

        Example:
            To get the current prices of a list of coins in USD, you can call:

            ```python
            coin_prices = coin_gecko.get_coin_price(ids=['bitcoin', 'ethereum'], vs_currencies=['usd'])
            ```

        Note:
            The `get_coin_price` method uses the `redis_cache` decorator to cache the result for a specified time (`ONE_MINUTE_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.simple.get_price(ids=ids, vs_currencies=vs_currencies)

        return result
    
    @redis_cache(ttl=ONE_HOUR_IN_SECONDS)
    @logger.instrument()
    def get_coin_data(self, id: str, cache_ttl: int = None) -> dict:
        """
        Retrieves detailed information about a specific coin.

        Args:
            id (str): The unique identifier of the coin.
            cache_ttl (int, optional): The time-to-live (in seconds) for the cached data. Defaults to None.

        Returns:
            dict: A dictionary containing the detailed information about the specified coin.

        Example:
            To get data for the coin with id 'bitcoin', you can call:

            ```python
            coin_data = coin_gecko.get_coin_data(id='bitcoin')
            ```

        Note:
            The `get_coin_data` 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.coins.get_id(id=id, localization=False, market_data=False, tickers=False, sparkline=False)
        logger.debug(f"Result: {result}")

        return result
    
    @redis_cache(ttl=ONE_MONTH_IN_SECONDS)
    @logger.instrument()
    def get_coin_category_data(self, params: dict = None, cache_ttl: int = None) -> dict:
        """
        Retrieves data for all available coin categories.

        Args:
            params (dict, optional): A dictionary containing optional parameters for the request. Defaults to None.
            cache_ttl (int, optional): The time-to-live (in seconds) for the cached data. Defaults to None.

        Returns:
            dict: A dictionary containing the data for all available coin categories.

        Example:
            To get data for all available coin categories, you can call:

            ```python
            category_data = coin_gecko.get_coin_category_data()
            ```

        Note:
            The `get_coin_category_data` 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.
        """
        params = params or {
            "order": "market_cap_desc",
        }

        result = self.cgc.categories.get_data(endpoint="coins/categories", params=params)
        logger.debug(f"Result: {result}")

        return { "categories": result}
    
    @redis_cache(ttl=ONE_MONTH_IN_SECONDS)
    @logger.instrument()
    def get_exchanges_list(self, cache_ttl: int = None) -> dict:
        """
        Retrieves a list of all available exchanges.

        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 exchanges.

        Example:
            To get a list of all available exchanges, you can call:

            ```python
            exchange_list = coin_gecko.get_exchanges_list()
            ```

        Note:
            The `get_exchanges_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.exchanges.get_list()
        logger.debug(f"Result: {result}")

        return { 'exchanges': result }
    
    @redis_cache(ttl=ONE_HOUR_IN_SECONDS)
    @logger.instrument()
    def get_exchange_data(self, id: str = None, cache_ttl: int = None) -> dict:
        """
        Retrieves detailed information about a specific exchange.

        Args:
            id (str, optional): The unique identifier of the exchange. Defaults to None.
            cache_ttl (int, optional): The time-to-live (in seconds) for the cached data. Defaults to None.

        Returns:
            dict: A dictionary containing the detailed information about the specified exchange.

        Raises:
            ValueError: If the `id` is not provided and `None` is passed as the argument.

        Example:
            To get data for the exchange with id 'binance', you can call:

            ```python
            exchange_data = coin_gecko.get_exchange_data(id='binance')
            ```

        Note:
            The `get_exchange_data` 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.
        """
        if id is None:
            raise ValueError("Exchange ID must be provided.")

        result = self.cgc.exchanges.get_id(id=id)
        logger.debug(f"Result: {result}")

        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 }