src/data_sources/cryptocompare.py

import os
from dotenv import load_dotenv
load_dotenv()

import httpx
from enum import Enum
from src.libs.logger import logger
from src.databases.redis import REDIS_CACHED
from src.libs.helper_functions import convert_to_snakecase
from src.libs.constants import (
    ONE_HOUR_IN_SECONDS,
    ONE_MONTH_IN_SECONDS,
    ONE_MINUTE_IN_SECONDS,
    ONE_QUARTER_IN_SECONDS,
    CRYPTO_COMPARE_API_BASE_URL,
    CRYPTO_COMPARE_ASSET_DATA_API_BASE_URL,
)

redis_cache = REDIS_CACHED

class CryptoCompareNewsSortOrder(Enum):
    LATEST = "latest"
    POPULAR = "popular"

    def __str__(self):
        return self.value
    
class CryptoCompareLanguages(Enum):
    English = "EN"
    French = "FR"
    Espanol = "ES"
    Turkish = "TR"
    Portuguese = "PT"

    def __str__(self):
        return self.value

class CryptoCompare:
    def __init__(self, base_url: str = None, asset_data_base_url: str = None) -> None:
        self.CRYPTO_COMPARE_API_BASE_URL = base_url or CRYPTO_COMPARE_API_BASE_URL
        self.CRYPTO_COMPARE_ASSET_DATA_API_BASE_URL = asset_data_base_url or CRYPTO_COMPARE_ASSET_DATA_API_BASE_URL

    # @redis_cache(ttl=ONE_MINUTE_IN_SECONDS)
    @logger.instrument()
    def get_all_coins(self) -> dict:
        url = f"{self.CRYPTO_COMPARE_API_BASE_URL}data/all/coinlist"
        logger.debug(url)
        params = {"api_key": os.getenv("CRYPTOCOMPARE_API_KEY")}
        logger.debug(params)
        headers = {"Content-type":"application/json; charset=UTF-8"}
        logger.debug(headers)

        try:
            with httpx.Client(timeout=30.0) as client:
                response = client.get(url, params=params, headers=headers)
                response.raise_for_status()  # Raise an exception if the request was unsuccessful
                return convert_to_snakecase(response.json())
        except httpx.HTTPError as e:
            print(f"An error occurred while making the request: {e}")
            return None
        
    @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:
        
        url = f"{self.CRYPTO_COMPARE_API_BASE_URL}data/pricemulti"
        # logger.debug(url)

        params = {
            "api_key": os.getenv("CRYPTOCOMPARE_API_KEY"),
            "fsyms": ",".join(ids),
            "tsyms": ",".join(vs_currencies)
        }
        # logger.debug(f"{params}")

        headers = {"Content-type":"application/json; charset=UTF-8"}
        # logger.debug(f"{headers}")

        try:
            with httpx.Client(timeout=30.0) as client:
                response = client.get(url, params=params, headers=headers)
                response.raise_for_status()  # Raise an exception if the request was unsuccessful
                return response.json()
        except httpx.HTTPError as e:
            # logger.debug(f"An error occurred while making the request: {e}")
            print(f"An error occurred while making the request: {e}")
            return None
        
    @redis_cache(ttl=ONE_QUARTER_IN_SECONDS)
    @logger.instrument()
    def get_overall_coin_data(self, symbol: str, cache_ttl: int = None) -> dict:
        
        url = f"{self.CRYPTO_COMPARE_ASSET_DATA_API_BASE_URL}data/by/symbol"
        logger.debug(f"Query URL: {url}")

        params = {
            "api_key": os.getenv("CRYPTOCOMPARE_API_KEY"),
            "asset_symbol": symbol,
        }
        logger.debug(f"Query params: {params}")

        headers = {"Content-type":"application/json; charset=UTF-8"}
        logger.debug(f"Query headers {headers}")

        try:
            with httpx.Client(timeout=30.0) as client:
                response = client.get(url, params=params, headers=headers)
                response.raise_for_status()  # Raise an exception if the request was unsuccessful
                return response.json()
        except httpx.HTTPError as e:
            logger.debug(f"An error occurred while making the request: {e}")
            # print(f"An error occurred while making the request: {e}")
            return None
        
    @redis_cache(ttl=ONE_QUARTER_IN_SECONDS)
    @logger.instrument()
    def get_news_categories(self, cache_ttl: int = None) -> dict:
        """
        Retrieves the news categories from the CryptoCompare API.
       
        Args:
            cache_ttl (int, optional): The time-to-live (TTL) for the cached response. Defaults to None.

        Returns:
            dict: A dictionary containing the news categories from the CryptoCompare API.

        Raises:
            httpx.HTTPError: If an error occurs while making the request to the CryptoCompare API.
        """
        url = f"{self.CRYPTO_COMPARE_API_BASE_URL}data/news/categories"
        logger.debug(f"Query URL: {url}")

        params = {
            "api_key": os.getenv("CRYPTOCOMPARE_API_KEY"),
        }
        logger.debug(f"Query params: {params}")

        headers = {"Content-type":"application/json; charset=UTF-8"}
        logger.debug(f"Query headers {headers}")

        try:
            with httpx.Client(timeout=30.0) as client:
                response = client.get(url, params=params, headers=headers)
                response.raise_for_status()  # Raise an exception if the request was unsuccessful
                return response.json()
        except httpx.HTTPError as e:
            logger.debug(f"An error occurred while making the request: {e}")
            # print(f"An error occurred while making the request: {e}")
            return None
        
    @redis_cache(ttl=ONE_HOUR_IN_SECONDS)
    @logger.instrument()
    def get_latest_news_articles(
        self,
        language: CryptoCompareLanguages = CryptoCompareLanguages.English,
        sort_order: CryptoCompareNewsSortOrder = CryptoCompareNewsSortOrder.LATEST,
        cache_ttl: int = None
    ) -> dict:
        """
        Retrieves the latest news articles from the CryptoCompare API.

        Args:
            language (CryptoCompareLanguages, optional): The language of the news articles. Defaults to English.
            sort_order (CryptoCompareNewsSortOrder, optional): The sort order of the news articles. Defaults to 'latest'.
            cache_ttl (int, optional): The time-to-live (TTL) for the cached response. Defaults to None.

        Returns:
            dict: A dictionary containing the latest news articles from the CryptoCompare API.

        Raises:
            httpx.HTTPError: If an error occurs while making the request to the CryptoCompare API.

        Example:
            # Get the latest news articles in English, sorted by popularity
            latest_news = crypto_compare_instance.get_latest_news_articles(language=CryptoCompareLanguages.English, sort_order=CryptoCompareNewsSortOrder.POPULAR)
        """
        url = f"{self.CRYPTO_COMPARE_API_BASE_URL}data/v2/news/"
        logger.debug(f"Query URL: {url}")

        params = {
            "api_key": os.getenv("CRYPTOCOMPARE_API_KEY"),
            "lang": language,
            "sortOrder": sort_order,
        }
        logger.debug(f"Query params: {params}")

        headers = {"Content-type":"application/json; charset=UTF-8"}
        logger.debug(f"Query headers {headers}")

        try:
            with httpx.Client(timeout=30.0) as client:
                response = client.get(url, params=params, headers=headers)
                response.raise_for_status()  # Raise an exception if the request was unsuccessful
                return response.json()
        except httpx.HTTPError as e:
            logger.debug(f"An error occurred while making the request: {e}")
            # print(f"An error occurred while making the request: {e}")
            return None
        
    @redis_cache(ttl=ONE_MONTH_IN_SECONDS)
    @logger.instrument()
    def get_all_exchanges_general_info(self, cache_ttl: int = None) -> dict:
        """
        Retrieves general information about all supported exchanges from the CryptoCompare API.

        Args:
            cache_ttl (int, optional): The time-to-live (TTL) for the cached response. Defaults to None.

        Returns:
            dict: A dictionary containing general information about all supported exchanges from the CryptoCompare API.

        Raises:
            httpx.HTTPError: If an error occurs while making the request to the CryptoCompare API.

        Example:
            # Get general information about all supported exchanges
            all_exchanges_info = crypto_compare_instance.get_all_exchanges_general_info()
        """
        url = f"{self.CRYPTO_COMPARE_API_BASE_URL}data/exchanges/general"
        logger.debug(f"Query URL: {url}")

        params = {
            "api_key": os.getenv("CRYPTOCOMPARE_API_KEY"),
        }
        logger.debug(f"Query params: {params}")

        headers = {"Content-type":"application/json; charset=UTF-8"}
        logger.debug(f"Query headers {headers}")

        try:
            with httpx.Client(timeout=30.0) as client:
                response = client.get(url, params=params, headers=headers)
                response.raise_for_status()  # Raise an exception if the request was unsuccessful
                return response.json()
        except httpx.HTTPError as e:
            logger.debug(f"An error occurred while making the request: {e}")
            # print(f"An error occurred while making the request: {e}")
            return None
        
    @redis_cache(ttl=ONE_MONTH_IN_SECONDS)
    @logger.instrument()
    def get_all_wallets_general_info(self, cache_ttl: int = None) -> dict:
        """
        Retrieves general information about all supported wallets from the CryptoCompare API.

        Args:
            cache_ttl (int, optional): The time-to-live (TTL) for the cached response. Defaults to None.

        Returns:
            dict: A dictionary containing general information about all supported wallets from the CryptoCompare API.

        Raises:
            httpx.HTTPError: If an error occurs while making the request to the CryptoCompare API.

        Example:
            # Get general information about all supported wallets
            all_wallets_info = crypto_compare_instance.get_all_wallets_general_info()
        """
        url = f"{self.CRYPTO_COMPARE_API_BASE_URL}data/wallets/general"
        logger.debug(f"Query URL: {url}")

        params = {
            "api_key": os.getenv("CRYPTOCOMPARE_API_KEY"),
        }
        logger.debug(f"Query params: {params}")

        headers = {"Content-type":"application/json; charset=UTF-8"}
        logger.debug(f"Query headers {headers}")

        try:
            with httpx.Client(timeout=30.0) as client:
                response = client.get(url, params=params, headers=headers)
                response.raise_for_status()  # Raise an exception if the request was unsuccessful
                return response.json()
        except httpx.HTTPError as e:
            logger.debug(f"An error occurred while making the request: {e}")
            # print(f"An error occurred while making the request: {e}")
            return None
        
    # @redis_cache(ttl=ONE_MONTH_IN_SECONDS)
    @logger.instrument()
    def get_top_tier_exchanges_list(
        self,
        top_tier: bool = False,
        cache_ttl: int = None
    ) -> dict:
        """
        Retrieves a list of top-tier exchanges from the CryptoCompare API.

        Args:
            top_tier (bool, optional): A boolean flag indicating whether to retrieve top-tier exchanges. Defaults to False.
            cache_ttl (int, optional): The time-to-live (TTL) for the cached response. Defaults to None.

        Returns:
            dict: A dictionary containing a list of top-tier exchanges from the CryptoCompare API.

        Raises:
            httpx.HTTPError: If an error occurs while making the request to the CryptoCompare API.

        Example:
            # Get a list of top-tier exchanges
            top_tier_exchanges = crypto_compare_instance.get_top_tier_exchanges_list(top_tier=True)
        """
        url = f"{self.CRYPTO_COMPARE_API_BASE_URL}data/v4/all/exchanges"
        logger.debug(f"Query URL: {url}")

        params = {
            "api_key": os.getenv("CRYPTOCOMPARE_API_KEY"),
            "topTier": 'true' if top_tier else 'false',
        }
        logger.debug(f"Query params: {params}")

        headers = {"Content-type":"application/json; charset=UTF-8"}
        logger.debug(f"Query headers {headers}")

        try:
            with httpx.Client(timeout=30.0) as client:
                response = client.get(url, params=params, headers=headers)
                response.raise_for_status()  # Raise an exception if the request was unsuccessful
                return response.json()
        except httpx.HTTPError as e:
            logger.debug(f"An error occurred while making the request: {e}")
            # print(f"An error occurred while making the request: {e}")
            return None