src/data_sources/dexscreener.py

from dotenv import load_dotenv

import httpx
from typing import List, Union
from src.databases.redis import REDIS_CACHED
from src.libs.constants import ONE_HOUR_IN_SECONDS, ONE_MINUTE_IN_SECONDS
from src.libs.constants import DEX_SCREENER_BASE_URL

load_dotenv()

redis_cache = REDIS_CACHED

class DexScreener:
    """
    A class for interacting with the Dex Screener API.

    Attributes:
        DEX_SCREENER_BASE_URL (str): The base URL for the Dex Screener API.

    Methods:
        __init__(self, base_url: str) -> None:
            Initialize the DexScreener class.

        get_pairs(self, chain_id: str, pair_addresses: str) -> dict:

        get_tokens(self, token_addresses: Union[str, List[str]]) -> dict:

        search_pairs(self, query: str) -> dict:
    """
    def __init__(self, base_url: str = None) -> None:
        self.DEX_SCREENER_BASE_URL = base_url or DEX_SCREENER_BASE_URL

    @redis_cache(ttl=ONE_MINUTE_IN_SECONDS)
    def get_pairs(self, chain_id: str, pair_addresses: str) -> dict:
        """
        This method is used to fetch pair data from Dex Screener API based on the provided chain_id and pair_addresses.

        Parameters:
        chain_id (str): The ID of the blockchain network.
        pair_addresses (str): The address(es) of the pair(s) on the blockchain.

        Returns:
        dict: A dictionary containing the JSON response from the API. Returns None if an error occurs.

        Raises:
        httpx.HTTPError: If an error occurs while making the request to the API.
        """
        url = f"{self.DEX_SCREENER_BASE_URL}pairs/{chain_id}/{pair_addresses}"
        try:
            response = httpx.get(url)
            response.raise_for_status()  # Raise an exception if the request was unsuccessful
            return 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)
    def get_tokens(self, token_addresses: Union[str, List[str]]) -> dict:
        """
        This method is used to fetch token data from Dex Screener API based on the provided token_addresses.

        Parameters:
        token_addresses (Union[str, List[str]]): The address(es) of the token(s) on the blockchain.
            This parameter can be a single address (str) or a list of addresses (List[str]).

        Returns:
        dict: A dictionary containing the JSON response from the API. Returns None if an error occurs.

        Raises:
        httpx.HTTPError: If an error occurs while making the request to the API.
        """
        if isinstance(token_addresses, list):
            token_addresses = ','.join(token_addresses)

        url = f"{self.DEX_SCREENER_BASE_URL}tokens/{token_addresses}"
        try:
            response = httpx.get(url)
            response.raise_for_status()  # Raise an exception if the request was unsuccessful
            return response.json()
        except httpx.HTTPError as e:
            print(f"An error occurred while making the request: {e}")
            return None
        
    @redis_cache(ttl=ONE_HOUR_IN_SECONDS)
    def search(self, query: str) -> dict:
        """
        This method is used to search for pairs matching the provided query from Dex Screener API.
        The query may include pair address, token address, token name, or token symbol.

        Parameters:
        query (str): The search query.

        Returns:
        dict: A dictionary containing the JSON response from the API. Returns None if an error occurs.

        Raises:
        httpx.HTTPError: If an error occurs while making the request to the API.
        """
        url = f"{self.DEX_SCREENER_BASE_URL}search/?q={query}"
        try:
            response = httpx.get(url)
            response.raise_for_status()  # Raise an exception if the request was unsuccessful
            return response.json()
        except httpx.HTTPError as e:
            print(f"An error occurred while making the request: {e}")
            return None