File size: 4,061 Bytes
9c400b9 383415a 9c400b9 383415a 9c400b9 383415a 9c400b9 383415a 9c400b9 383415a 9c400b9 383415a 9c400b9 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 |
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
|